catman/yt-mls
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Patched arbitrary file system access vulnerability and visual changes

Browse Source
This commit is contained in:
Maximilian Wagner
2023-08-04 20:45:25 +02:00
parent 9953ef8aab
commit 59bacdbe6d
8 changed files with 128 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
backend.py
View File

@@ -2,8 +2,10 @@ import threading
import yt_dlp as ydl
import os
import zipfile
from datetime import datetime
from datetime import datetime
from urllib.request import urlopen
from urllib.error import URLError
from base64 import b64encode
from threading import Thread
@@ -21,9 +23,17 @@ def enqueue_download(url, update=False, ext='mp3'):
def process_general(url, ext, update=False):
# get current time and put in list to be displayed on /index
# get current time
current_time = datetime.now().time()
running_downloads.append([url, str(current_time.hour) + ':' + str(current_time.minute)])
# parse hour and minute
hour = str(current_time.hour)
hour = hour if len(hour) > 1 else '0' + hour
minute = str(current_time.minute)
minute = minute if len(minute) > 1 else '0' + hour
# add url and time to list of queued downloads
queued_downloads.append([url, hour + ':' + minute])
# wait for previous thread to finish if not first / only in list
current_thread = threading.current_thread()
@@ -45,7 +55,7 @@ def process_general(url, ext, update=False):
process_download(url, ext, parent, query, current_thread)
try:
running_downloads.pop(0)
queued_downloads.pop(0)
except IndexError:
print('*** IndexError: download could not be removed from list of running downloads. ***')
@@ -324,7 +334,7 @@ def downloads_path() -> str:
# updates zip in or creates new zip of given directory
def zip_folder(full_rel_path):
def zip_folder(full_rel_path) -> tuple[str, str]:
# get playlist name
parent = full_rel_path.split('/')
for folder in parent:
@@ -350,7 +360,7 @@ def zip_folder(full_rel_path):
# create archive
zipfile.ZipFile(downloads_path() + full_rel_path + filename, 'w')
# Open the existing zip file in append mode
# add remaining files to zip
with zipfile.ZipFile(downloads_path() + full_rel_path + filename, 'a') as existing_zip:
file_list = existing_zip.namelist()
file_list = [e[len(parent)+1:] for e in file_list]
@@ -358,9 +368,31 @@ def zip_folder(full_rel_path):
for entry in os.scandir(downloads_path() + full_rel_path):
if entry.is_file() and not entry.name.endswith('.zip') and entry.name not in file_list:
# Add the file to the zip, preserving the directory structure
existing_zip.write(entry.path, arcname=parent + '\\' + entry.name)
existing_zip.write(entry.path, arcname=parent + '/' + entry.name)
return filename
return full_rel_path, filename
def zip_folder_not_in_directory(zip_full_rel_path):
video_not_in_directory = query_db_threaded('SELECT path, name, ext FROM video '
'INNER JOIN collection ON video.id = collection.video '
'WHERE NOT path = playlist')
# get full path to downloads directory
downloads_folder = downloads_path()
# add remaining files to zip
with zipfile.ZipFile(downloads_folder + zip_full_rel_path, 'a') as existing_zip:
file_list_zip = existing_zip.namelist()
file_list_files = [e.split('/')[-1] for e in file_list_zip]
file_list_folder = file_list_zip[0].split('/')[0]
for video in video_not_in_directory:
file_name = video[1] + video[2]
if file_name not in file_list_files:
existing_zip.write(downloads_folder + video[0] + file_name, arcname=file_list_folder + '/' + file_name)
return
def directory_contains_zip(full_rel_path):
@@ -368,3 +400,11 @@ def directory_contains_zip(full_rel_path):
if file.name.endswith('.zip'):
return file.name
return ''
def internet_available(target='http://www.youtube.com'):
try:
urlopen(target)
return True
except URLError:
return False

2
file_cache.py
View File

@@ -2,4 +2,4 @@ ids = []
titles = []
urls = []
thread_queue = []
running_downloads = []
queued_downloads = []

53
frontend.py
View File

@@ -4,7 +4,7 @@ from __future__ import unicode_literals
from flask import Blueprint, request, render_template, flash, send_from_directory, send_file
from forms.download import DownloadForm
from backend import zip_folder, downloads_path, enqueue_download
from backend import zip_folder, zip_folder_not_in_directory, downloads_path, enqueue_download, internet_available
from db_tools import query_db
from file_cache import *
@@ -14,9 +14,9 @@ frontend = Blueprint('frontend', __name__)
# index has a list of running downloads
@frontend.route('/', methods=['GET'])
def index():
if not running_downloads:
if not queued_downloads:
flash('Currently, no downloads are running.', 'primary')
return render_template('index.html', running_downloads=running_downloads, titles=titles, urls=urls, amount=len(urls))
return render_template('index.html', running_downloads=queued_downloads, titles=titles, urls=urls, amount=len(urls))
@frontend.route('/downloader', methods=['GET', 'POST'])
@@ -28,7 +28,7 @@ def downloader():
url = str(form.url.data)
ext = str(form.ext.data)
# check if valid link
# check if high likelihood of being a valid link
# this tool should technically work with other platforms but that is not tested
# since KeyErrors are to be expected in backend.process_download(url), it's blocked here
# you are invited to test and adjust the code for other platforms
@@ -39,30 +39,44 @@ def downloader():
valid_link = True if url == 'None' else False # if url is empty, don't show error
return render_template('downloader.html', form=form, ytLink=valid_link, amount=len(urls))
if not internet_available():
flash('No internet connection available.', 'danger')
return render_template('flash-message.html')
# kick off download process
enqueue_download(url, ext=ext)
# show download start confirmation
flash('Download enqueued and will finish in background.', 'primary')
return render_template('feedback-simple.html', amount=len(urls))
return render_template('flash-message.html')
# downloads a single file
@frontend.route('/download/<path:file_path>', methods=['GET'])
def download(file_path):
def download(file_path: str):
# if the path does not end with a slash, a single file is requested
if '.' in file_path:
file_folder = ''.join([x if x not in file_path.split('/')[-1] else '' for x in file_path.split('/')])
video = query_db('SELECT path, name, ext FROM video WHERE name = :name AND path = :path',
{'name': file_path.split('/')[-1].split('.')[0], 'path': file_folder + '\\' if file_folder else ''},
True)
return send_from_directory(
downloads_path(),
file_path
downloads_path() + video['path'],
video['name'] + video['ext']
)
# else a directory is requested
else:
zip_path, zip_name = zip_folder(file_path)
print(zip_path, zip_name)
zip_folder_not_in_directory(zip_path + zip_name)
# zip and send
return send_from_directory(
downloads_path(),
file_path + zip_folder(file_path)
downloads_path() + zip_path,
zip_name
)
@@ -76,7 +90,7 @@ def update(url_rowid):
# show download start confirmation
flash('Update enqueued and will finish in background.', 'primary')
return render_template('feedback-simple.html', titles=titles, urls=urls, amount=len(urls))
return render_template('flash-message.html', titles=titles, urls=urls, amount=len(urls))
@frontend.route('/library', methods=['GET'])
@@ -93,11 +107,20 @@ def library():
@frontend.route('/library-playlist', methods=['GET'])
def library_playlist():
playlist = request.args.get('playlist', None)
videos = query_db('SELECT video.name, video.ext, video.path FROM video LEFT JOIN collection ON video.id = '
'collection.video LEFT JOIN playlist ON collection.playlist=playlist.folder WHERE '
'playlist.ROWID = :playlist',
videos = query_db('SELECT video.name, video.ext, video.path FROM video '
'LEFT JOIN collection ON video.id = collection.video '
'LEFT JOIN playlist ON collection.playlist=playlist.folder '
'WHERE playlist.ROWID = :playlist',
{'playlist': playlist})
return render_template('collection.html', videos=videos)
# get playlist path since could be empty in some entries
folder = ''
for video in videos:
if len(video['path']) > 0:
folder = video['path']
break
return render_template('collection.html', videos=videos, folder=folder)
@frontend.route('/player', methods=['GET'])

2
templates/collection.html
View File

@@ -25,7 +25,7 @@
</div>
</div>
<div class="container" style="padding: 1.5%">
<form action="/download/{{ videos[0]['path'] }}">
<form action="/download/{{ folder }}">
<input type="submit" class="btn btn-primary float-end" value="Download all"/>
</form>
</div>

13
templates/downloader.html
View File

@@ -12,13 +12,14 @@
{{ form.url.label(class_="form-label") }}
</div>
<div class="row" style="width: fit-content">
<div class="col">
{{ form.url(class_="form-control") }}
{{ form.url(class_="form-control") }}
</div>
<br>
<div class="row" style="width: fit-content">
<div class="col-md-auto">
{{ form.ext(class_="form-select") }}
</div>
<div class="col">
{{ form.ext(class_="form-select") }}
</div>
<div class="col" style="width: fit-content">
<div class="col-md-auto">
{{ form.submit(class_="btn btn-primary") }}
</div>
</div>

31
templates/feedback-simple.html
View File

@@ -1,31 +0,0 @@
{%- extends "base.html" %}
{% block content %}
{{ super() }}
<div class="container">
{% if titles %}
<div class="container">
<table id="videos" class="table">
<thead>
<tr>
<th scope="col" class="text-center">New Title(s)</th>
<th scope="col" class="text-center">URL</th>
</tr>
</thead>
<tbody>
{% for i in range(amount) %}
<tr>
<td class="text-center">{{ titles[i] }}</td>
<td class="text-center"> <a href="{{ urls[i] }}" target="_blank">Link</a></td>
</tr>
{% endfor %}
</tbody>
</table>
<form target="/start-download"
<input type="submit"
</div>
{% endif %}
</div>
{%- endblock %}

1
templates/flash-message.html Normal file
View File

@@ -0,0 +1 @@
{%- extends "base.html" %}

74
templates/index.html
View File

@@ -6,28 +6,22 @@
{% if running_downloads %}
<div class="card">
<div class="card-body">
<ul class="list-group list-group-flush">
<li class="list-group-item">
<div class="row">
<table id="videos" class="table">
<thead>
<tr>
<th scope="col" class="text-center">Queue</th>
<th scope="col" class="text-center">Started at</th>
</tr>
</thead>
<tbody>
{% for entry in running_downloads %}
<tr>
<td class="text-center"><a href="{{ entry[0] }}" target="_blank">{{ entry[0] }}</a></td>
<td class="text-center">{{ entry[1] }}</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
</li>
</ul>
<table class="table">
<thead>
<tr>
<th scope="col" class="text-center">Queue</th>
<th scope="col" class="text-center">Started at</th>
</tr>
</thead>
<tbody>
{% for entry in running_downloads %}
<tr>
<td class="text-center"><a href="{{ entry[0] }}" target="_blank">{{ entry[0] }}</a></td>
<td class="text-center">{{ entry[1] }}</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
</div>
@@ -35,26 +29,22 @@
{% if titles %}
<div class="card">
<ul class="list-group list-group-flush">
<li class="list-group-item">
<div class="row">
<table id="videos" class="table">
<thead>
<tr>
<th scope="col" class="text-center">Currently processing</th>
</tr>
</thead>
<tbody>
{% for i in range(amount) %}
<tr>
<td class="text-center"><a href="{{ urls[i] }}" target="_blank">{{ titles[i] }}</a></td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
</li>
</ul>
<div class="card-body">
<table class="table">
<thead>
<tr>
<th scope="col" class="text-center">Currently processing</th>
</tr>
</thead>
<tbody>
{% for i in range(amount) %}
<tr>
<td class="text-center"><a href="{{ urls[i] }}" target="_blank">{{ titles[i] }}</a></td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
</div>
{% endif %}
{% endif %}