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 <noreply@anthropic.com>
This commit is contained in:
ubuntu
2026-08-30 23:19:54 +00:00
co-authored by Claude Sonnet 5
parent 21d40c932e
commit 55cefe47f4
+27 -4
View File
@@ -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: