hzhang
58f23ddcb8
Critical: - backup: prevent Zip Slip path traversal and zip bombs in restore/convert via safe_extract(); serialize get_backup() with backup_lock and always restore CWD so concurrent requests can't corrupt the os.chdir state - app: only enable the Werkzeug debugger/reloader when ENVIRONMENT=dev; always init rate limits (also under WSGI), not just under __main__ - apikey: fix create_key never committing (session.commit -> commit()), validate roles against an allowlist, and fix revoke_key/update_last_used operating on detached instances so revocation actually persists - env_provider: redact DB_PASSWORD and SESSION_SECRET_KEY in summerize() High: - markdown: filter private/protected docs for non-admins in the listing, get_home, get_index and search endpoints (was an anonymous data leak); escape LIKE metacharacters and cap search results - webhooks: validate target URL to block SSRF (loopback/private/link-local/ metadata IPs), disable redirects, safely parse additional_header - auth: validate JWT issuer and require exp/iat; add timeout to JWKS fetch; harden Authorization header parsing against malformed values - log: require admin for GET /api/log and auth for POST; bound entry size Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
54 lines
1.8 KiB
Python
54 lines
1.8 KiB
Python
#api/log.py
|
|
from flask import Blueprint, jsonify, request
|
|
from api import require_auth
|
|
from db import get_db
|
|
from db.models.Log import Log
|
|
from db.utils import insert_log
|
|
|
|
logs_bp = Blueprint('log', __name__, url_prefix='/api/log')
|
|
|
|
# Bound per-entry size so an authenticated-but-low-trust caller can't bloat
|
|
# the log table with multi-megabyte payloads.
|
|
_MAX_LOG_MESSAGE_LEN = 16 * 1024
|
|
|
|
@logs_bp.route('/', methods=['GET'])
|
|
@require_auth(roles=['admin'])
|
|
def get_logs():
|
|
level = request.args.get('level')
|
|
application = request.args.get('application')
|
|
page = int(request.args.get('page', 1))
|
|
per_page = int(request.args.get('per_page', 10))
|
|
|
|
with get_db() as session:
|
|
query = session.query(Log)
|
|
if level:
|
|
query = query.filter(Log.level == level)
|
|
if application:
|
|
query = query.filter(Log.application == application)
|
|
total_logs = query.count()
|
|
logs = query.order_by(Log.timestamp.desc()).offset((page - 1)*per_page).limit(per_page).all()
|
|
return jsonify({
|
|
"total": total_logs,
|
|
"page": page,
|
|
"per_page": per_page,
|
|
"logs": [log.to_dict() for log in logs]
|
|
})
|
|
|
|
@logs_bp.route('/', methods=['POST'])
|
|
@require_auth()
|
|
def create_log():
|
|
data = request.get_json(silent=True)
|
|
if not data:
|
|
return jsonify({"error": "invalid or missing JSON body"}), 400
|
|
required_fields = ['level', 'message']
|
|
for field in required_fields:
|
|
if field not in data:
|
|
return jsonify({"error": f"missing {field} in request"}), 400
|
|
level = str(data.get('level'))[:64]
|
|
message = str(data.get('message'))[:_MAX_LOG_MESSAGE_LEN]
|
|
application = "frontend"
|
|
extra = data.get('extra', None)
|
|
log_entry = Log(level=level, message=message, application=application, extra=extra)
|
|
insert_log(log_entry)
|
|
return jsonify({"message": "log created"}), 201
|