From 55cefe47f48095d29277468b9f7eed8dcaa3d327 Mon Sep 17 00:00:00 2001 From: ubuntu Date: Sun, 30 Aug 2026 23:19:54 +0000 Subject: [PATCH] Fix path traversal in file-serving and file-writing endpoints playlist_download, download_epg, upload_snapshot, and get_logo all built filesystem paths by concatenating a caller-supplied string (filename/id_channel/id_device) with no validation, so a '../' payload could read or write files outside the intended directory. upload_snapshot was the most exploitable: fully attacker-controlled bytes written to an attacker-chosen path with no prior lookup gating it. Add a safe_join() helper that resolves the path and rejects anything that escapes the base directory, and use it at all four call sites. Co-Authored-By: Claude Sonnet 5 --- server.py | 31 +++++++++++++++++++++++++++---- 1 file changed, 27 insertions(+), 4 deletions(-) diff --git a/server.py b/server.py index 6a4d222..c5a7054 100644 --- a/server.py +++ b/server.py @@ -99,6 +99,16 @@ loader = Loader('') # Generic functions +def safe_join(base_dir, user_filename): + """Resolve user_filename under base_dir, rejecting anything that would escape it + (e.g. '../../etc/passwd') via path separators or symlinks.""" + base_dir = os.path.realpath(base_dir) + candidate = os.path.realpath(os.path.join(base_dir, user_filename or '')) + if candidate != base_dir and not candidate.startswith(base_dir + os.sep): + raise ValueError('invalid filename') + return candidate + + def findInfo(tag, s): end = '"' if (s.find(tag)>-1): @@ -416,7 +426,10 @@ async def index(request): # Access to playlist file from the web async def playlist_download(request): data = dict(request.query) - file_path = '/home/playlists_download/'+data['filename'] + try: + file_path = safe_join('/home/playlists_download', data['filename']) + except ValueError: + raise web.HTTPBadRequest(text='Invalid filename') if os.path.exists(file_path): return web.FileResponse(file_path) else: @@ -871,7 +884,11 @@ async def upload_snapshot(request): if not id_device or not file_bytes: return web.json_response({'status': 'error', 'message': 'missing id_device or file'}, status=400) - destination = DEVICE_SNAPSHOT_DIR / f'{id_device}.jpg' + try: + destination = safe_join(str(DEVICE_SNAPSHOT_DIR), f'{id_device}.jpg') + except ValueError: + return web.json_response({'status': 'error', 'message': 'invalid id_device'}, status=400) + with open(destination, 'wb') as output_file: output_file.write(file_bytes) @@ -1077,7 +1094,10 @@ async def get_logo(request): data = dict(request.query) id_channel = data['id_channel'] id_playlist = data['id_playlist'] - logo_channel_file_path = LOGO_DIR+'/'+id_channel+'.jpg' + try: + logo_channel_file_path = safe_join(LOGO_DIR, id_channel + '.jpg') + except ValueError: + raise web.HTTPBadRequest(text='Invalid id_channel') collection_channels = db_iptv['channels_'+id_playlist] d = collection_channels.find_one({ 'id_channel':id_channel}) url_logo = d['urlLogo'] @@ -1372,7 +1392,10 @@ async def download_epg(request): data = dict(request.query) url = data['url'] filename = data['filename'] - filepath = os.path.join(DOWNLOAD_EPG_DIR, filename) + try: + filepath = safe_join(DOWNLOAD_EPG_DIR, filename) + except ValueError: + raise web.HTTPBadRequest(text='Invalid filename') print('download file :', url, flush=True) response = requests.get(url, timeout=10) with open(filepath, 'wb') as f: