445 lines
19 KiB
Python
445 lines
19 KiB
Python
import logging
|
|
import os
|
|
import secrets
|
|
from datetime import datetime
|
|
from functools import wraps
|
|
|
|
from flask import Flask, jsonify, render_template, request, session
|
|
from werkzeug.middleware.proxy_fix import ProxyFix
|
|
import click
|
|
|
|
import database
|
|
from security import (
|
|
CATEGORIES,
|
|
RateLimiter,
|
|
ValidationError,
|
|
csrf_token,
|
|
integer,
|
|
iso_datetime,
|
|
require_json,
|
|
text,
|
|
valid_csrf,
|
|
validate_credentials,
|
|
validate_event,
|
|
validate_feed,
|
|
validate_public_url,
|
|
)
|
|
|
|
|
|
logging.basicConfig(level=os.environ.get('LOG_LEVEL', 'INFO'))
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _env_bool(name, default):
|
|
value = os.environ.get(name)
|
|
if value is None:
|
|
return default
|
|
if value.lower() not in {'true', 'false'}:
|
|
raise RuntimeError(f'{name} deve ser true ou false.')
|
|
return value.lower() == 'true'
|
|
|
|
|
|
def _secret(environment, supplied=None):
|
|
value = supplied or os.environ.get('FLASK_SECRET_KEY')
|
|
if environment == 'production':
|
|
if (not value or len(value) < 32 or len(set(value)) < 8
|
|
or value in {'super-secret-key-calendar-study', 'changeme'}):
|
|
raise RuntimeError('FLASK_SECRET_KEY forte (mínimo de 32 caracteres) é obrigatória em produção.')
|
|
return value
|
|
return value or secrets.token_hex(32)
|
|
|
|
|
|
def create_app(test_config=None):
|
|
environment = os.environ.get('APP_ENV', 'production').lower()
|
|
if environment not in {'production', 'development', 'test'}:
|
|
raise RuntimeError('APP_ENV deve ser production, development ou test.')
|
|
supplied_secret = (test_config or {}).get('SECRET_KEY')
|
|
trusted_hosts_value = os.environ.get('TRUSTED_HOSTS', '')
|
|
trusted_hosts = [host.strip() for host in trusted_hosts_value.split(',') if host.strip()]
|
|
if environment == 'production' and not trusted_hosts and not (test_config or {}).get('TRUSTED_HOSTS'):
|
|
raise RuntimeError('TRUSTED_HOSTS é obrigatório em produção.')
|
|
application = Flask(__name__, static_folder='static', template_folder='templates')
|
|
application.config.update(
|
|
SECRET_KEY=_secret(environment, supplied_secret),
|
|
MAX_CONTENT_LENGTH=int(os.environ.get('MAX_CONTENT_LENGTH', str(64 * 1024))),
|
|
MAX_FORM_MEMORY_SIZE=int(os.environ.get('MAX_FORM_MEMORY_SIZE', str(16 * 1024))),
|
|
MAX_FORM_PARTS=20,
|
|
SESSION_COOKIE_HTTPONLY=True,
|
|
SESSION_COOKIE_SECURE=_env_bool('SESSION_COOKIE_SECURE', environment == 'production'),
|
|
SESSION_COOKIE_SAMESITE=os.environ.get('SESSION_COOKIE_SAMESITE', 'Lax'),
|
|
SESSION_COOKIE_NAME='__Host-studycalendar' if environment == 'production' else 'studycalendar',
|
|
SESSION_COOKIE_PATH='/',
|
|
SESSION_REFRESH_EACH_REQUEST=False,
|
|
PERMANENT_SESSION_LIFETIME=86400,
|
|
DATABASE_PATH=os.environ.get(
|
|
'DATABASE_PATH', os.path.join(application.root_path, 'calendario.db')
|
|
),
|
|
ICS_HTTPS_ONLY=_env_bool('ICS_HTTPS_ONLY', True),
|
|
ENABLE_HSTS=_env_bool('ENABLE_HSTS', environment == 'production'),
|
|
RATELIMIT_ENABLED=True,
|
|
ALLOW_REGISTRATION=_env_bool('ALLOW_REGISTRATION', environment != 'production'),
|
|
MAX_FEEDS_PER_USER=int(os.environ.get('MAX_FEEDS_PER_USER', '20')),
|
|
MAX_MANUAL_EVENTS_PER_USER=int(os.environ.get('MAX_MANUAL_EVENTS_PER_USER', '10000')),
|
|
MAX_NOTIFICATION_RULES_PER_USER=int(os.environ.get('MAX_NOTIFICATION_RULES_PER_USER', '20')),
|
|
MAX_EVENT_RESULTS=int(os.environ.get('MAX_EVENT_RESULTS', '5000')),
|
|
TRUSTED_HOSTS=trusted_hosts or None,
|
|
)
|
|
if test_config:
|
|
application.config.update(test_config)
|
|
|
|
bounded_config = {
|
|
'MAX_FEEDS_PER_USER': (1, 100),
|
|
'MAX_MANUAL_EVENTS_PER_USER': (1, 100000),
|
|
'MAX_NOTIFICATION_RULES_PER_USER': (1, 100),
|
|
'MAX_EVENT_RESULTS': (1, 10000),
|
|
}
|
|
for name, (minimum, maximum) in bounded_config.items():
|
|
if not minimum <= application.config[name] <= maximum:
|
|
raise RuntimeError(f'{name} deve estar entre {minimum} e {maximum}.')
|
|
|
|
same_site = application.config['SESSION_COOKIE_SAMESITE']
|
|
if same_site not in {'Strict', 'Lax', 'None'}:
|
|
raise RuntimeError('SESSION_COOKIE_SAMESITE deve ser Strict, Lax ou None.')
|
|
if same_site == 'None' and not application.config['SESSION_COOKIE_SECURE']:
|
|
raise RuntimeError('SESSION_COOKIE_SAMESITE=None exige SESSION_COOKIE_SECURE=true.')
|
|
if environment == 'production' and not application.config['SESSION_COOKIE_SECURE']:
|
|
raise RuntimeError('SESSION_COOKIE_SECURE deve permanecer true em produção.')
|
|
|
|
trusted_proxies = int(os.environ.get('TRUSTED_PROXY_COUNT', '0'))
|
|
if trusted_proxies < 0 or trusted_proxies > 5:
|
|
raise RuntimeError('TRUSTED_PROXY_COUNT deve estar entre 0 e 5.')
|
|
if trusted_proxies:
|
|
application.wsgi_app = ProxyFix(
|
|
application.wsgi_app, x_for=trusted_proxies, x_proto=trusted_proxies,
|
|
x_host=trusted_proxies, x_port=trusted_proxies
|
|
)
|
|
|
|
database.configure(application.config['DATABASE_PATH'])
|
|
database.init_db()
|
|
limiter = RateLimiter()
|
|
|
|
def success_response(data=None, message='Operação realizada com sucesso.', status_code=200):
|
|
return jsonify({'success': True, 'data': data, 'message': message}), status_code
|
|
|
|
def error_response(message='Ocorreu um erro.', status_code=400, headers=None):
|
|
response = jsonify({'success': False, 'data': None, 'message': message})
|
|
return (response, status_code, headers or {})
|
|
|
|
def login_required(function):
|
|
@wraps(function)
|
|
def decorated(*args, **kwargs):
|
|
user_id = session.get('user_id')
|
|
session_version = session.get('session_version')
|
|
if user_id is None or session_version is None:
|
|
return error_response('Não autorizado.', 401)
|
|
if database.get_user_session_version(user_id) != session_version:
|
|
session.clear()
|
|
return error_response('Sessão expirada.', 401)
|
|
return function(*args, **kwargs)
|
|
return decorated
|
|
|
|
def not_found_if_false(found, message):
|
|
return None if found else error_response(message, 404)
|
|
|
|
@application.before_request
|
|
def protect_request():
|
|
if request.path.startswith('/api/') and not valid_csrf(request.host_url):
|
|
return error_response('Token CSRF ausente ou origem inválida.', 403)
|
|
if not application.config['RATELIMIT_ENABLED'] or not request.path.startswith('/api/'):
|
|
return None
|
|
identity = session.get('user_id') or request.remote_addr or 'unknown'
|
|
limit = 120
|
|
bucket = 'global'
|
|
if request.path in {'/api/auth/login', '/api/auth/register'}:
|
|
identity = request.remote_addr or 'unknown'
|
|
limit, bucket = 10, 'auth'
|
|
elif '/sync' in request.path or (request.path == '/api/feeds' and request.method == 'POST'):
|
|
limit, bucket = 15, 'feed'
|
|
allowed, retry = limiter.check((bucket, identity), limit)
|
|
if not allowed:
|
|
return error_response('Muitas requisições. Tente novamente depois.', 429,
|
|
{'Retry-After': str(retry)})
|
|
return None
|
|
|
|
@application.after_request
|
|
def secure_headers(response):
|
|
response.headers['Content-Security-Policy'] = (
|
|
"default-src 'self'; base-uri 'none'; object-src 'none'; frame-ancestors 'none'; "
|
|
"form-action 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; "
|
|
"img-src 'self' data:; connect-src 'self'; manifest-src 'self'"
|
|
)
|
|
response.headers['X-Content-Type-Options'] = 'nosniff'
|
|
response.headers['Referrer-Policy'] = 'no-referrer'
|
|
response.headers['X-Frame-Options'] = 'DENY'
|
|
response.headers['Permissions-Policy'] = (
|
|
'camera=(), microphone=(), geolocation=(), payment=(), usb=()'
|
|
)
|
|
if application.config['ENABLE_HSTS']:
|
|
response.headers['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains'
|
|
if request.path.startswith('/api/'):
|
|
response.headers['Cache-Control'] = 'no-store, max-age=0'
|
|
response.headers['Pragma'] = 'no-cache'
|
|
return response
|
|
|
|
@application.errorhandler(ValidationError)
|
|
def validation_error(error):
|
|
return error_response(str(error), 400)
|
|
|
|
@application.errorhandler(413)
|
|
def too_large(_error):
|
|
return error_response('Corpo da requisição excede o limite permitido.', 413)
|
|
|
|
@application.errorhandler(404)
|
|
def route_not_found(_error):
|
|
if request.path.startswith('/api/'):
|
|
return error_response('Recurso não encontrado.', 404)
|
|
return _error
|
|
|
|
@application.errorhandler(500)
|
|
def internal_error(error):
|
|
if request.path.startswith('/api/'):
|
|
logger.error('Erro interno na API (path=%s, tipo=%s).', request.path,
|
|
type(getattr(error, 'original_exception', error)).__name__)
|
|
return error_response('Erro interno do servidor.', 500)
|
|
return error
|
|
|
|
@application.route('/')
|
|
def index():
|
|
return render_template('index.html')
|
|
|
|
@application.cli.command('create-user')
|
|
@click.option('--username', prompt=True)
|
|
@click.option('--display-name', prompt='Nome de exibição')
|
|
@click.password_option(confirmation_prompt=True)
|
|
def create_user_command(username, display_name, password):
|
|
"""Cria um usuário sem habilitar cadastro público."""
|
|
try:
|
|
normalized, validated_password, validated_name = validate_credentials({
|
|
'username': username,
|
|
'password': password,
|
|
'display_name': display_name,
|
|
}, True)
|
|
except ValidationError as error:
|
|
raise click.ClickException(str(error)) from error
|
|
if not database.create_user(normalized, validated_password, validated_name):
|
|
raise click.ClickException('Não foi possível criar o usuário.')
|
|
click.echo('Usuário criado.')
|
|
|
|
@application.get('/api/auth/csrf')
|
|
def get_csrf():
|
|
return success_response({
|
|
'csrf_token': csrf_token(),
|
|
'registration_enabled': application.config['ALLOW_REGISTRATION'],
|
|
})
|
|
|
|
@application.post('/api/auth/register')
|
|
def register():
|
|
if not application.config['ALLOW_REGISTRATION']:
|
|
return error_response('Cadastro público desativado.', 403)
|
|
username, password, display_name = validate_credentials(require_json(), True)
|
|
user_id = database.create_user(username, password, display_name)
|
|
if not user_id:
|
|
return error_response('Não foi possível criar a conta com os dados informados.', 409)
|
|
request_csrf = session.get('csrf_token')
|
|
session.clear()
|
|
session['user_id'] = user_id
|
|
session['username'] = username
|
|
session['session_version'] = database.get_user_session_version(user_id)
|
|
session['csrf_token'] = request_csrf or csrf_token()
|
|
return success_response({'id': user_id, 'username': username}, 'Usuário registrado com sucesso.')
|
|
|
|
@application.post('/api/auth/login')
|
|
def login():
|
|
username, password, _ = validate_credentials(require_json())
|
|
user = database.verify_user_password(username, password)
|
|
if not user:
|
|
return error_response('Usuário ou senha inválidos.', 401)
|
|
request_csrf = session.get('csrf_token')
|
|
session.clear()
|
|
session['user_id'] = user['id']
|
|
session['username'] = user['username']
|
|
session['session_version'] = user['session_version']
|
|
session['csrf_token'] = request_csrf or csrf_token()
|
|
return success_response(
|
|
{'id': user['id'], 'username': user['username'], 'display_name': user['display_name']},
|
|
'Login realizado.'
|
|
)
|
|
|
|
@application.post('/api/auth/logout')
|
|
def logout():
|
|
user_id = session.get('user_id')
|
|
if user_id is not None:
|
|
database.revoke_user_sessions(user_id)
|
|
session.clear()
|
|
return success_response(message='Logout realizado.')
|
|
|
|
@application.get('/api/auth/me')
|
|
@login_required
|
|
def get_me():
|
|
user = database.get_user_by_id(session['user_id'])
|
|
return success_response(user) if user else error_response('Usuário não encontrado.', 404)
|
|
|
|
@application.get('/api/events')
|
|
@login_required
|
|
def list_events():
|
|
start = iso_datetime(request.args.get('start'), 'start')
|
|
end = iso_datetime(request.args.get('end'), 'end')
|
|
if start and end:
|
|
start_value = datetime.fromisoformat(start)
|
|
end_value = datetime.fromisoformat(end)
|
|
if bool(start_value.tzinfo) != bool(end_value.tzinfo) or start_value > end_value:
|
|
raise ValidationError('end não pode ser anterior a start e deve usar o mesmo fuso.')
|
|
category = request.args.get('category')
|
|
if category and category not in CATEGORIES | {'todas'}:
|
|
raise ValidationError('category inválida.')
|
|
search = request.args.get('search')
|
|
if search is not None and len(search) > 100:
|
|
raise ValidationError('search excede 100 caracteres.')
|
|
return success_response(database.get_events(
|
|
session['user_id'], start, end, category, search,
|
|
limit=application.config['MAX_EVENT_RESULTS']
|
|
))
|
|
|
|
@application.post('/api/events')
|
|
@login_required
|
|
def create_event():
|
|
values = validate_event(require_json())
|
|
event_id = database.add_event(
|
|
user_id=session['user_id'], max_count=application.config['MAX_MANUAL_EVENTS_PER_USER'],
|
|
**values
|
|
)
|
|
if not event_id:
|
|
return error_response('Limite de eventos manuais atingido.', 409)
|
|
return success_response({'id': event_id}, 'Evento criado com sucesso.')
|
|
|
|
@application.put('/api/events/<int:event_id>')
|
|
@login_required
|
|
def update_event_route(event_id):
|
|
values = validate_event(require_json())
|
|
if not database.update_event(event_id=event_id, user_id=session['user_id'], **values):
|
|
return error_response('Evento não encontrado.', 404)
|
|
return success_response(message='Evento atualizado com sucesso.')
|
|
|
|
@application.delete('/api/events/<int:event_id>')
|
|
@login_required
|
|
def delete_event_route(event_id):
|
|
missing = not_found_if_false(database.delete_event(event_id, session['user_id']), 'Evento não encontrado.')
|
|
return missing or success_response(message='Evento excluído com sucesso.')
|
|
|
|
@application.patch('/api/events/<int:event_id>/toggle')
|
|
@login_required
|
|
def toggle_event(event_id):
|
|
if not database.toggle_event_completed(event_id, session['user_id']):
|
|
return error_response('Evento não encontrado.', 404)
|
|
return success_response(message='Status do evento alterado.')
|
|
|
|
@application.get('/api/feeds')
|
|
@login_required
|
|
def list_feeds():
|
|
return success_response(database.get_feeds(session['user_id']))
|
|
|
|
@application.post('/api/feeds')
|
|
@login_required
|
|
def add_feed_route():
|
|
values = validate_feed(require_json(), application.config['ICS_HTTPS_ONLY'])
|
|
validate_public_url(values['url'], application.config['ICS_HTTPS_ONLY'])
|
|
feed_id = database.add_feed(
|
|
user_id=session['user_id'], max_count=application.config['MAX_FEEDS_PER_USER'],
|
|
**values
|
|
)
|
|
if not feed_id:
|
|
return error_response('Feed duplicado ou limite de feeds atingido.', 409)
|
|
return success_response(
|
|
{'id': feed_id, 'sync_queued': True}, 'Feed adicionado; sincronização agendada.', 202
|
|
)
|
|
|
|
@application.delete('/api/feeds/<int:feed_id>')
|
|
@login_required
|
|
def delete_feed_route(feed_id):
|
|
missing = not_found_if_false(database.delete_feed(feed_id, session['user_id']), 'Feed não encontrado.')
|
|
return missing or success_response(message='Feed excluído com sucesso.')
|
|
|
|
@application.post('/api/feeds/<int:feed_id>/sync')
|
|
@login_required
|
|
def sync_feed_route(feed_id):
|
|
if not database.request_feed_sync(feed_id, session['user_id']):
|
|
return error_response('Feed não encontrado.', 404)
|
|
return success_response(message='Sincronização agendada.', status_code=202)
|
|
|
|
@application.post('/api/feeds/sync-all')
|
|
@login_required
|
|
def sync_all_feeds_route():
|
|
feeds = database.get_feeds(session['user_id'])
|
|
for feed in feeds:
|
|
database.request_feed_sync(feed['id'], session['user_id'])
|
|
return success_response(message='Sincronizações agendadas.', status_code=202)
|
|
|
|
@application.get('/api/notifications/upcoming')
|
|
@login_required
|
|
def get_upcoming_notifications():
|
|
return success_response(database.get_upcoming_notifications(session['user_id']))
|
|
|
|
@application.post('/api/notifications/mark-notified')
|
|
@login_required
|
|
def mark_notified():
|
|
data = require_json()
|
|
unknown = set(data) - {'event_id', 'rule_id'}
|
|
if unknown:
|
|
raise ValidationError(f"Campos desconhecidos: {', '.join(sorted(unknown))}.")
|
|
event_id = integer(data, 'event_id', minimum=1, maximum=2 ** 63 - 1)
|
|
rule_id = integer(data, 'rule_id', minimum=1, maximum=2 ** 63 - 1)
|
|
if not database.mark_event_notified(event_id, rule_id, session['user_id']):
|
|
return error_response('Evento ou regra não encontrado.', 404)
|
|
return success_response(message='Evento marcado como notificado para a regra.')
|
|
|
|
@application.get('/api/notifications/rules')
|
|
@login_required
|
|
def get_notification_rules_route():
|
|
return success_response(database.get_notification_rules(session['user_id']))
|
|
|
|
@application.post('/api/notifications/rules')
|
|
@login_required
|
|
def add_notification_rule_route():
|
|
data = require_json()
|
|
unknown = set(data) - {'minutes_before', 'label'}
|
|
if unknown:
|
|
raise ValidationError(f"Campos desconhecidos: {', '.join(sorted(unknown))}.")
|
|
minutes = integer(data, 'minutes_before', minimum=1, maximum=10080, default=30)
|
|
label = text(data, 'label', required=True, minimum=1, maximum=100)
|
|
rule_id = database.add_notification_rule(
|
|
session['user_id'], minutes, label,
|
|
max_count=application.config['MAX_NOTIFICATION_RULES_PER_USER']
|
|
)
|
|
if not rule_id:
|
|
return error_response('Regra duplicada ou limite de regras atingido.', 409)
|
|
return success_response({'id': rule_id}, 'Regra criada.')
|
|
|
|
@application.delete('/api/notifications/rules/<int:rule_id>')
|
|
@login_required
|
|
def delete_notification_rule_route(rule_id):
|
|
if not database.delete_notification_rule(rule_id, session['user_id']):
|
|
return error_response('Regra não encontrada.', 404)
|
|
return success_response(message='Regra removida.')
|
|
|
|
@application.patch('/api/notifications/rules/<int:rule_id>/toggle')
|
|
@login_required
|
|
def toggle_notification_rule_route(rule_id):
|
|
if not database.toggle_notification_rule(rule_id, session['user_id']):
|
|
return error_response('Regra não encontrada.', 404)
|
|
return success_response(message='Status da regra alterado.')
|
|
|
|
@application.get('/api/stats')
|
|
@login_required
|
|
def get_stats_route():
|
|
return success_response(database.get_stats(session['user_id']))
|
|
|
|
return application
|
|
|
|
|
|
app = create_app()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
if os.environ.get('APP_ENV') != 'development':
|
|
raise RuntimeError('O servidor embutido só pode ser usado com APP_ENV=development.')
|
|
app.run(host=os.environ.get('DEV_HOST', '127.0.0.1'), port=int(os.environ.get('PORT', '5000')), debug=False)
|