-
Notifications
You must be signed in to change notification settings - Fork 297
/
__main__.py
184 lines (146 loc) · 6.27 KB
/
__main__.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
import os
import signal
import argparse
from flask import Flask, render_template, send_file, redirect, request, send_from_directory, url_for, abort
from flask_httpauth import HTTPBasicAuth
from werkzeug.utils import secure_filename
from werkzeug.security import generate_password_hash, check_password_hash
from werkzeug.serving import run_simple
from updog.utils.path import is_valid_subpath, get_parent_directory, process_files
from updog.utils.output import error, info, warn, success
from updog import version as VERSION
def read_write_directory(directory):
if os.path.exists(directory):
if os.access(directory, os.W_OK and os.R_OK):
return directory
else:
error('The output is not readable and/or writable')
else:
error('The specified directory does not exist')
def parse_arguments():
parser = argparse.ArgumentParser(prog='updog')
cwd = os.getcwd()
parser.add_argument('-d', '--directory', metavar='DIRECTORY', type=read_write_directory, default=cwd,
help='Root directory\n'
'[Default=.]')
parser.add_argument('-p', '--port', type=int, default=9090,
help='Port to serve [Default=9090]')
parser.add_argument('--password', type=str, default='', help='Use a password to access the page. (No username)')
parser.add_argument('--ssl', action='store_true', help='Use an encrypted connection')
parser.add_argument('--version', action='version', version='%(prog)s v'+VERSION)
args = parser.parse_args()
# Normalize the path
args.directory = os.path.abspath(args.directory)
return args
def main():
args = parse_arguments()
app = Flask(__name__)
auth = HTTPBasicAuth()
global base_directory
base_directory = args.directory
# Deal with Favicon requests
@app.route('/favicon.ico')
def favicon():
return send_from_directory(os.path.join(app.root_path, 'static'),
'images/favicon.ico', mimetype='image/vnd.microsoft.icon')
############################################
# File Browsing and Download Functionality #
############################################
@app.route('/', defaults={'path': None})
@app.route('/<path:path>')
@auth.login_required
def home(path):
# If there is a path parameter and it is valid
if path and is_valid_subpath(path, base_directory):
# Take off the trailing '/'
path = os.path.normpath(path)
requested_path = os.path.join(base_directory, path)
# If directory
if os.path.isdir(requested_path):
back = get_parent_directory(requested_path, base_directory)
is_subdirectory = True
# If file
elif os.path.isfile(requested_path):
# Check if the view flag is set
if request.args.get('view') is None:
send_as_attachment = True
else:
send_as_attachment = False
# Check if file extension
(filename, extension) = os.path.splitext(requested_path)
if extension == '':
mimetype = 'text/plain'
else:
mimetype = None
try:
return send_file(requested_path, mimetype=mimetype, as_attachment=send_as_attachment)
except PermissionError:
abort(403, 'Read Permission Denied: ' + requested_path)
else:
# Root home configuration
is_subdirectory = False
requested_path = base_directory
back = ''
if os.path.exists(requested_path):
# Read the files
try:
directory_files = process_files(os.scandir(requested_path), base_directory)
except PermissionError:
abort(403, 'Read Permission Denied: ' + requested_path)
return render_template('home.html', files=directory_files, back=back,
directory=requested_path, is_subdirectory=is_subdirectory, version=VERSION)
else:
return redirect('/')
#############################
# File Upload Functionality #
#############################
@app.route('/upload', methods=['POST'])
@auth.login_required
def upload():
if request.method == 'POST':
# No file part - needs to check before accessing the files['file']
if 'file' not in request.files:
return redirect(request.referrer)
path = request.form['path']
# Prevent file upload to paths outside of base directory
if not is_valid_subpath(path, base_directory) or path == '':
return redirect(request.referrer)
for file in request.files.getlist('file'):
# No filename attached
if file.filename == '':
return redirect(request.referrer)
# Assuming all is good, process and save out the file
# TODO:
# - Add support for overwriting
if file:
filename = secure_filename(file.filename)
full_path = os.path.join(path, filename)
try:
file.save(full_path)
except PermissionError:
abort(403, 'Write Permission Denied: ' + full_path)
return redirect(request.referrer)
# Password functionality is without username
users = {
'': generate_password_hash(args.password)
}
@auth.verify_password
def verify_password(username, password):
if args.password:
if username in users:
return check_password_hash(users.get(username), password)
return False
else:
return True
# Inform user before server goes up
success('Serving {}...'.format(args.directory, args.port))
def handler(signal, frame):
print()
error('Exiting!')
signal.signal(signal.SIGINT, handler)
ssl_context = None
if args.ssl:
ssl_context = 'adhoc'
run_simple("0.0.0.0", int(args.port), app, ssl_context=ssl_context)
if __name__ == '__main__':
main()