201 lines
6.4 KiB
Python
201 lines
6.4 KiB
Python
# pylint: disable=C0114
|
|
import os
|
|
import re
|
|
import datetime
|
|
from flask import Flask, request, jsonify, send_from_directory, render_template
|
|
from flasgger import Swagger
|
|
|
|
app = Flask(__name__, template_folder='templates')
|
|
swagger = Swagger(app)
|
|
app.config['UPLOAD_DIRECTORY'] = os.environ.get('UPLOAD_DIRECTORY', '/uploads')
|
|
app.config['MAX_CONTENT_LENGTH'] = int(os.environ.get('MAX_CONTENT_LENGTH', '5')) * 1024 * 1024 # in MB # pylint: disable=C0301
|
|
app.config['ENABLE_WEBSERVER'] = os.getenv('ENABLE_WEBSERVER', 'True').lower() == 'true'
|
|
|
|
|
|
VALID_FILENAME_REGEX = r'^[a-zA-Z0-9\-_\.]+$'
|
|
|
|
AUTH_TOKEN = os.environ.get('AUTH_TOKEN', 'myuploadtoken')
|
|
|
|
def is_valid_filename(filename): # pylint: disable=C0116
|
|
return bool(re.match(VALID_FILENAME_REGEX, filename))
|
|
|
|
if app.config['ENABLE_WEBSERVER']:
|
|
@app.route('/', methods=['GET'])
|
|
def file_list():
|
|
"""
|
|
Endpoint for displaying a list of files in the upload directory.
|
|
"""
|
|
files = []
|
|
for filename in os.listdir(app.config['UPLOAD_DIRECTORY']):
|
|
file_path = os.path.join(app.config['UPLOAD_DIRECTORY'], filename)
|
|
if os.path.isfile(file_path):
|
|
stats = os.stat(file_path)
|
|
size = stats.st_size
|
|
last_modified = datetime.datetime.fromtimestamp(stats.st_mtime).strftime('%Y-%m-%d %H:%M:%S') # pylint: disable=C0301
|
|
files.append({
|
|
'name': filename,
|
|
'size': size,
|
|
'last_modified': last_modified
|
|
})
|
|
|
|
return render_template('file_list.html', files=files)
|
|
|
|
@app.route('/health', methods=['GET'])
|
|
def health_check():
|
|
"""
|
|
Endpoint for health check.
|
|
---
|
|
responses:
|
|
200:
|
|
description: OK
|
|
"""
|
|
return 'OK'
|
|
|
|
@app.route('/upload', methods=['POST'])
|
|
def upload_file():
|
|
"""
|
|
Endpoint for uploading files.
|
|
Filename can only contain alphanumeric characters, hyphens, underscores, and periods.
|
|
If the filename is the same as an existing file, this file will be overwritten.
|
|
---
|
|
parameters:
|
|
- name: file
|
|
in: formData
|
|
type: file
|
|
required: true
|
|
description: The file to upload.
|
|
- name: token
|
|
in: header
|
|
type: string
|
|
required: true
|
|
description: Authentication token.
|
|
responses:
|
|
200:
|
|
description: File uploaded successfully.
|
|
400:
|
|
description: Bad request.
|
|
401:
|
|
description: Unauthorized.
|
|
"""
|
|
if 'file' not in request.files:
|
|
return jsonify({'error': 'No file part in the request'}), 400
|
|
|
|
if 'token' not in request.headers:
|
|
return jsonify({'error': 'No token supplied'}), 401
|
|
|
|
if request.headers['token'] != AUTH_TOKEN:
|
|
return jsonify({'error': 'Invalid token supplied'}), 401
|
|
|
|
file = request.files['file']
|
|
if file.filename == '':
|
|
return jsonify({'error': 'No file selected for upload'}), 400
|
|
|
|
if not is_valid_filename(file.filename):
|
|
return jsonify({'error': 'Invalid filename. Only alphanumeric characters, hyphens, underscores, and periods are allowed.'}), 400 # pylint: disable=C0301
|
|
|
|
filename = file.filename
|
|
file.save(os.path.join(app.config['UPLOAD_DIRECTORY'], filename))
|
|
return jsonify({'success': 'File \'{}\' successfully uploaded'.format(filename)}) # pylint: disable=C0209
|
|
|
|
@app.route('/download/<filename>', methods=['GET'])
|
|
def download_file(filename):
|
|
"""
|
|
Endpoint for downloading files.
|
|
---
|
|
parameters:
|
|
- name: filename
|
|
in: path
|
|
type: string
|
|
required: true
|
|
description: The name of the file to download.
|
|
responses:
|
|
200:
|
|
description: File downloaded successfully.
|
|
404:
|
|
description: File not found.
|
|
"""
|
|
try:
|
|
return send_from_directory(app.config['UPLOAD_DIRECTORY'], filename)
|
|
except FileNotFoundError:
|
|
return jsonify({'error': 'File \'{}\' not found'}), 404
|
|
|
|
@app.route('/delete/<filename>', methods=['DELETE'])
|
|
def delete_file(filename):
|
|
"""
|
|
Endpoint for deleting files.
|
|
---
|
|
parameters:
|
|
- name: filename
|
|
in: path
|
|
type: string
|
|
required: true
|
|
description: The name of the file to delete.
|
|
- name: token
|
|
in: header
|
|
type: string
|
|
required: true
|
|
description: Authentication token.
|
|
responses:
|
|
200:
|
|
description: File deleted successfully.
|
|
401:
|
|
description: Unauthorized.
|
|
404:
|
|
description: File not found.
|
|
"""
|
|
if 'token' not in request.headers:
|
|
return jsonify({'error': 'No token supplied'}), 401
|
|
|
|
if request.headers['token'] != AUTH_TOKEN:
|
|
return jsonify({'error': 'Invalid token supplied'}), 401
|
|
|
|
file_path = os.path.join(app.config['UPLOAD_DIRECTORY'], filename)
|
|
if not os.path.isfile(file_path):
|
|
return jsonify({'error': 'File not found'}), 404
|
|
|
|
os.remove(file_path)
|
|
return jsonify({'success': 'File \'{}\' successfully deleted'.format(filename)}) # pylint: disable=C0209
|
|
|
|
|
|
@app.route('/list', methods=['GET'])
|
|
def list_files():
|
|
"""
|
|
Endpoint for listing files.
|
|
---
|
|
parameters:
|
|
- name: token
|
|
in: header
|
|
type: string
|
|
required: true
|
|
description: Authentication token.
|
|
responses:
|
|
200:
|
|
description: Listed files successfully.
|
|
400:
|
|
description: Bad request.
|
|
401:
|
|
description: Unauthorized.
|
|
"""
|
|
if 'token' not in request.headers:
|
|
return jsonify({'error': 'No token supplied'}), 401
|
|
|
|
if request.headers['token'] != AUTH_TOKEN:
|
|
return jsonify({'error': 'Invalid token supplied'}), 401
|
|
|
|
files = []
|
|
for filename in os.listdir(app.config['UPLOAD_DIRECTORY']):
|
|
file_path = os.path.join(app.config['UPLOAD_DIRECTORY'], filename)
|
|
if os.path.isfile(file_path):
|
|
stats = os.stat(file_path)
|
|
size = stats.st_size
|
|
last_modified = datetime.datetime.fromtimestamp(stats.st_mtime).strftime('%Y-%m-%d %H:%M:%S') # pylint: disable=C0301
|
|
files.append({
|
|
'name': filename,
|
|
'size': size,
|
|
'last_modified': last_modified
|
|
})
|
|
|
|
return jsonify({'files': files})
|
|
|
|
if __name__ == '__main__':
|
|
app.run(debug=True, host='0.0.0.0', port=int(os.environ.get('PORT', 5000)))
|