Security hardening: fix RCE, auth and SSRF issues

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>
This commit is contained in:
h z
2026-05-16 16:12:43 +01:00
parent 1f4ca52a10
commit 58f23ddcb8
8 changed files with 225 additions and 35 deletions

View File

@@ -215,6 +215,40 @@ logger = logging.getLogger(__name__)
backup_bp = Blueprint('backup', __name__, url_prefix='/api/backup')
# Upper bounds to defend against zip bombs in uploaded backups.
_MAX_ARCHIVE_MEMBERS = 50000
_MAX_UNCOMPRESSED_BYTES = 1 * 1024 * 1024 * 1024 # 1 GiB
def safe_extract(zip_ref, dest_dir):
"""
Safely extract a zip archive into dest_dir.
Prevents "Zip Slip" path traversal (entries like ``../../etc/x`` or
absolute paths) and rejects zip bombs by capping the member count and
total uncompressed size. Backups are admin-uploaded but may originate
from an untrusted source (e.g. the /convert endpoint ingests foreign
backups), so the contents must be treated as hostile.
"""
dest_root = os.path.realpath(dest_dir)
members = zip_ref.infolist()
if len(members) > _MAX_ARCHIVE_MEMBERS:
raise ValueError("backup archive has too many entries")
total = 0
for member in members:
total += member.file_size
if total > _MAX_UNCOMPRESSED_BYTES:
raise ValueError("backup archive uncompressed size exceeds limit")
target = os.path.realpath(os.path.join(dest_root, member.filename))
if target != dest_root and not target.startswith(dest_root + os.sep):
raise ValueError(f"unsafe path in backup archive: {member.filename}")
zip_ref.extractall(dest_root)
def check_and_convert_backup_version(backup_dir):
"""
Check the backup version and convert it if necessary.
@@ -280,7 +314,7 @@ def convert_backup_endpoint():
uploaded_file.save(zip_path)
with zipfile.ZipFile(zip_path, 'r') as zip_ref:
zip_ref.extractall(backup_dir)
safe_extract(zip_ref, backup_dir)
success, error_response = check_and_convert_backup_version(backup_dir)
if not success:
@@ -331,7 +365,12 @@ def get_backup():
Response Codes:
- 200: Backup created successfully
- 500: Failed to create backup
- 429: Another backup operation is in progress
"""
if not backup_lock.acquire(blocking=False):
return jsonify({"error": "Another backup operation is in progress. Please try again later."}), 429
original_cwd = os.getcwd()
try:
if os.path.exists('Root'):
shutil.rmtree('Root')
@@ -370,6 +409,11 @@ def get_backup():
except Exception as e:
logger.error(f"Failed to get backup: {e}")
return jsonify({"error": "failed to get backup"}), 500
finally:
# os.chdir is process-global; always restore CWD so a failure
# mid-traverse can't corrupt later requests.
os.chdir(original_cwd)
backup_lock.release()
def create_and_cd(path_name):
@@ -552,7 +596,7 @@ def load_backup():
uploaded_file.save(zip_path)
with zipfile.ZipFile(zip_path, 'r') as zip_ref:
zip_ref.extractall(temp_dir)
safe_extract(zip_ref, temp_dir)
root_dir = os.path.join(temp_dir, "Root")
if not os.path.exists(root_dir):