else/SimpleServer/simpleserver.py

463 lines
13 KiB
Python
Raw Normal View History

2020-01-20 05:19:45 +00:00
import argparse
import cgi
import http.cookies
2015-09-03 21:09:35 +00:00
import http.server
2020-01-20 05:19:45 +00:00
import math
2016-07-05 07:24:08 +00:00
import mimetypes
2015-10-03 23:29:52 +00:00
import os
2016-07-20 03:31:47 +00:00
import pathlib
2016-07-05 07:24:08 +00:00
import random
2016-07-20 03:31:47 +00:00
import socketserver
2016-07-05 07:24:08 +00:00
import sys
2019-06-12 05:29:46 +00:00
import threading
2016-07-20 03:31:47 +00:00
import types
2019-06-12 05:29:46 +00:00
import urllib.parse
import zipstream
2016-07-05 07:24:08 +00:00
2016-12-25 03:18:23 +00:00
# pip install voussoirkit
2016-12-02 06:37:07 +00:00
from voussoirkit import bytestring
from voussoirkit import pathclass
from voussoirkit import ratelimiter
2015-09-03 21:09:35 +00:00
2016-07-20 03:31:47 +00:00
FILE_READ_CHUNK = bytestring.MIBIBYTE
2016-08-18 01:24:38 +00:00
RATELIMITER = ratelimiter.Ratelimiter(16 * bytestring.MIBIBYTE)
2016-07-20 03:31:47 +00:00
2016-07-05 07:24:08 +00:00
OPENDIR_TEMPLATE = '''
<html>
<body>
<meta charset="UTF-8">
2019-06-12 05:29:46 +00:00
<meta name="viewport" content="width=device-width, initial-scale=1.0"/>
2016-07-05 07:24:08 +00:00
<style type="text/css">Body {{font-family:Consolas}}</style>
2019-06-12 05:29:46 +00:00
<table style="width: 100%">
{table_rows}
2016-07-05 07:24:08 +00:00
</table>
</body>
</html>
'''
2020-01-20 05:19:45 +00:00
PASSWORD_PROMPT_HTML = '''
<html>
<body>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0"/>
<style type="text/css">Body {{font-family:Consolas}}</style>
<form action="/password" method="post">
<input type="text" name="password" placeholder="password" autocomplete="off"/>
2020-01-20 05:19:45 +00:00
<input type="hidden" name="goto" value="{goto}"/>
<input type="submit" value="Submit"/>
</form>
</body>
</html>
'''
2019-06-12 05:29:46 +00:00
ROOT_DIRECTORY = pathclass.Path(os.getcwd())
2020-10-10 15:43:12 +00:00
TOKEN_COOKIE_NAME = 'simpleserver_token'
2015-10-03 23:29:52 +00:00
2020-10-10 15:44:56 +00:00
# SERVER ###########################################################################################
2015-09-03 21:09:35 +00:00
class RequestHandler(http.server.BaseHTTPRequestHandler):
2020-10-10 15:44:56 +00:00
def __init__(self, *args, password=None, accepted_tokens=None, **kwargs):
2020-01-20 05:19:45 +00:00
self.accepted_tokens = accepted_tokens
2020-10-10 15:44:56 +00:00
self.password = password
2020-01-20 05:19:45 +00:00
super().__init__(*args, **kwargs)
@property
def auth_token(self):
cookie = self.headers.get('Cookie')
if not cookie:
return None
cookie = http.cookies.SimpleCookie(cookie)
token = cookie.get(TOKEN_COOKIE_NAME)
if not token:
return None
return token
2020-01-20 05:19:45 +00:00
def check_password(self, attempt):
if self.password is None:
return True
if attempt == self.password:
return True
return False
def check_has_password(self):
if self.password is None:
return True
if self.headers.get('password', None) == self.password:
return True
if self.accepted_tokens is not None and self.auth_token in self.accepted_tokens:
return True
2020-01-20 05:19:45 +00:00
return False
2016-07-20 03:31:47 +00:00
def write(self, data):
if isinstance(data, str):
data = data.encode('utf-8')
if isinstance(data, types.GeneratorType):
for chunk in data:
self.wfile.write(chunk)
2016-08-18 01:24:38 +00:00
RATELIMITER.limit(len(chunk))
2016-07-20 03:31:47 +00:00
else:
self.wfile.write(data)
2015-09-03 21:09:35 +00:00
def do_GET(self):
2020-01-20 05:19:45 +00:00
if not self.check_has_password():
self.send_response(401)
self.end_headers()
self.write(PASSWORD_PROMPT_HTML.format(goto=self.path))
return
2019-06-12 05:29:46 +00:00
path = url_to_path(self.path)
2016-07-10 04:38:49 +00:00
if self.send_path_validation_error(path):
2015-10-03 23:29:52 +00:00
return
2016-07-05 07:24:08 +00:00
2016-07-20 03:31:47 +00:00
range_min = None
range_max = None
status_code = 200
headers = {}
2016-07-05 07:24:08 +00:00
2016-07-20 03:31:47 +00:00
if path.is_file:
file_size = path.size
if 'range' in self.headers:
desired_range = self.headers['range']
desired_range = desired_range.lower()
desired_range = desired_range.split('bytes=')[-1]
helper = lambda x: int(x) if x and x.isdigit() else None
if '-' in desired_range:
(desired_min, desired_max) = desired_range.split('-')
#print('desire', desired_min, desired_max)
range_min = helper(desired_min)
range_max = helper(desired_max)
else:
range_min = helper(desired_range)
if range_min is None:
range_min = 0
if range_max is None:
range_max = file_size
# because ranges are 0 indexed
range_max = min(range_max, file_size - 1)
range_min = max(range_min, 0)
status_code = 206
range_header = 'bytes {min}-{max}/{outof}'.format(
min=range_min,
max=range_max,
outof=file_size,
)
headers['Content-Range'] = range_header
headers['Accept-Ranges'] = 'bytes'
content_length = (range_max - range_min) + 1
else:
content_length = file_size
headers['Content-length'] = content_length
2019-06-12 05:29:46 +00:00
if path.is_file:
headers['Content-type'] = mimetypes.guess_type(path.absolute_path)[0]
response = read_filebytes(path, range_min=range_min, range_max=range_max)
elif path.is_dir:
headers['Content-type'] = 'text/html'
response = generate_opendir(path)
response = response.encode('utf-8')
elif self.path.endswith('.zip'):
path = url_to_path(self.path.rsplit('.zip', 1)[0])
headers['Content-type'] = 'application/octet-stream'
2020-01-20 05:19:45 +00:00
download_as = urllib.parse.quote(path.basename)
download_as += '.zip'
headers['Content-Disposition'] = f'attachment; filename*=UTF-8\'\'{download_as}'
2019-06-12 05:29:46 +00:00
response = zip_directory(path)
response = iter(response)
# response = (print(chunk) or chunk for chunk in response)
else:
status_code = 404
self.send_error(status_code)
response = bytes()
2016-07-10 04:38:49 +00:00
2016-07-20 03:31:47 +00:00
self.send_response(status_code)
for (key, value) in headers.items():
self.send_header(key, value)
2016-07-05 07:24:08 +00:00
2016-07-10 04:38:49 +00:00
self.end_headers()
2019-06-12 05:29:46 +00:00
self.write(response)
2015-09-03 21:09:35 +00:00
2016-07-10 04:38:49 +00:00
def do_HEAD(self):
2020-01-20 05:19:45 +00:00
if not self.check_has_password():
self.send_response(401)
self.end_headers()
return
2019-06-12 05:29:46 +00:00
path = url_to_path(self.path)
2016-07-10 04:38:49 +00:00
if self.send_path_validation_error(path):
return
2016-07-20 03:31:47 +00:00
status_code = 200
2016-08-03 02:06:59 +00:00
self.send_response(status_code)
2016-07-10 04:38:49 +00:00
if path.is_dir:
mime = 'text/html'
else:
2016-07-20 03:31:47 +00:00
mime = mimetypes.guess_type(path.absolute_path)[0]
self.send_header('Content-length', path.size)
2016-07-10 04:38:49 +00:00
if mime is not None:
self.send_header('Content-type', mime)
self.end_headers()
2020-01-20 05:19:45 +00:00
def do_POST(self):
2020-10-10 15:44:56 +00:00
(ctype, pdict) = cgi.parse_header(self.headers.get('content-type'))
2020-01-20 05:19:45 +00:00
if ctype == 'multipart/form-data':
form = cgi.parse_multipart(self.rfile, pdict)
elif ctype == 'application/x-www-form-urlencoded':
length = int(self.headers.get('content-length'))
form = urllib.parse.parse_qs(self.rfile.read(length), keep_blank_values=1)
else:
form = {}
if self.path == '/password':
attempt = form.get(b'password')[0].decode('utf-8')
goto = form.get(b'goto')[0].decode('utf-8')
if self.check_password(attempt):
cookie = http.cookies.SimpleCookie()
token = random_hex(32)
2020-10-10 15:43:12 +00:00
cookie[TOKEN_COOKIE_NAME] = token
2020-01-20 05:19:45 +00:00
self.accepted_tokens.add(token)
self.send_response(302)
self.send_header('Set-Cookie', cookie.output(header='', sep=''))
self.send_header('Location', goto)
else:
self.send_response(401)
2020-10-10 15:44:56 +00:00
elif not self.check_has_password():
self.send_response(401)
self.end_headers()
return
2020-01-20 05:19:45 +00:00
else:
self.send_response(400)
self.end_headers()
2016-07-10 04:38:49 +00:00
def send_path_validation_error(self, path):
2019-06-12 05:29:46 +00:00
if not allowed(path):
2016-07-20 03:31:47 +00:00
self.send_error(403, 'Stop that!')
2016-07-10 04:38:49 +00:00
return True
return False
2020-10-10 15:44:56 +00:00
class SimpleServer:
def __init__(self, port, password):
self.port = port
self.password = password
self.accepted_tokens = set()
def make_request_handler(self, *args, **kwargs):
return RequestHandler(
password=self.password,
accepted_tokens=self.accepted_tokens,
*args,
**kwargs,
)
def start(self):
server = http.server.ThreadingHTTPServer(('0.0.0.0', self.port), self.make_request_handler)
print(f'Server starting on {self.port}')
try:
server.serve_forever()
except KeyboardInterrupt:
print('Goodbye.')
server.shutdown()
# HELPERS ##########################################################################################
2016-07-20 03:31:47 +00:00
2019-06-12 05:29:46 +00:00
def allowed(path):
2019-12-10 20:46:45 +00:00
return path == ROOT_DIRECTORY or path in ROOT_DIRECTORY
2019-06-12 05:29:46 +00:00
2019-12-10 20:46:45 +00:00
def atag(path, display_name=None):
2019-06-12 05:29:46 +00:00
path.correct_case()
if display_name is None:
display_name = path.basename
if path.is_dir:
# Folder emoji
icon = '\U0001F4C1'
else:
# Diamond emoji
#icon = '\U0001F48E'
icon = '\U0001F381'
#print('anchor', path)
if display_name.endswith('.placeholder'):
a = '<a>{icon} {display}</a>'
else:
a = '<a href="{full}">{icon} {display}</a>'
a = a.format(
full=path_to_url(path),
icon=icon,
display=display_name,
)
return a
2016-07-05 07:24:08 +00:00
def generate_opendir(path):
2016-07-10 04:38:49 +00:00
#print('Listdir:', path)
# This places directories above files, each ordered alphabetically
2019-06-12 05:29:46 +00:00
try:
items = path.listdir()
except FileNotFoundError:
items = []
2016-07-05 07:24:08 +00:00
directories = []
files = []
2019-06-12 05:29:46 +00:00
for item in sorted(items, key=lambda p: p.basename.lower()):
if item.is_dir:
2016-07-05 07:24:08 +00:00
directories.append(item)
else:
2019-12-10 20:46:45 +00:00
if item.basename.lower() == 'thumbs.db':
continue
if item.basename.lower() == 'desktop.ini':
continue
2016-07-05 07:24:08 +00:00
files.append(item)
2016-07-10 04:38:49 +00:00
items = directories + files
2019-06-12 05:29:46 +00:00
table_rows = []
shaded = False
2016-07-20 03:31:47 +00:00
2019-06-12 05:29:46 +00:00
if path.absolute_path == ROOT_DIRECTORY.absolute_path:
2016-07-20 03:31:47 +00:00
# This is different than a permission check, we're seeing if they're
# actually at the top, in which case they don't need an up button.
2019-12-10 20:46:45 +00:00
entry = table_row(path, display_name='.', shaded=shaded)
table_rows.append(entry)
shaded = not shaded
2016-07-20 03:31:47 +00:00
else:
2019-06-12 05:29:46 +00:00
entry = table_row(path.parent, display_name='up', shaded=shaded)
table_rows.append(entry)
shaded = not shaded
2016-07-10 04:38:49 +00:00
for item in items:
2019-06-12 05:29:46 +00:00
entry = table_row(item, shaded=shaded)
table_rows.append(entry)
2016-07-05 07:24:08 +00:00
shaded = not shaded
2016-07-10 04:38:49 +00:00
2020-01-20 05:19:45 +00:00
if len(items) > 0:
entry = table_row(path.replace_extension('.zip'), display_name='zip', shaded=shaded)
shaded = not shaded
table_rows.append(entry)
2019-06-12 05:29:46 +00:00
table_rows = '\n'.join(table_rows)
text = OPENDIR_TEMPLATE.format(table_rows=table_rows)
2016-07-05 07:24:08 +00:00
return text
def generate_random_filename(original_filename='', length=8):
import random
2016-07-20 03:31:47 +00:00
bits = length * 44
2016-07-05 07:24:08 +00:00
bits = random.getrandbits(bits)
identifier = '{:x}'.format(bits).rjust(length, '0')
return identifier
2020-01-20 05:19:45 +00:00
def random_hex(length=12):
randbytes = os.urandom(math.ceil(length / 2))
token = ''.join('{:02x}'.format(x) for x in randbytes)
token = token[:length]
return token
2019-06-12 05:29:46 +00:00
def read_filebytes(path, range_min=None, range_max=None):
#print(path)
if range_min is None:
range_min = 0
if range_max is None:
range_max = path.size
range_span = range_max - range_min
#print('read span', range_min, range_max, range_span)
2020-09-21 01:29:36 +00:00
f = path.open('rb')
2019-06-12 05:29:46 +00:00
f.seek(range_min)
sent_amount = 0
while sent_amount < range_span:
chunk = f.read(FILE_READ_CHUNK)
if len(chunk) == 0:
break
yield chunk
sent_amount += len(chunk)
#print('I read', len(fr))
f.close()
def table_row(path, display_name=None, shaded=False):
form = '<tr style="background-color:#{bg}"><td style="">{anchor}</td><td>{size}</td></tr>'
if path.is_file:
size = bytestring.bytestring(path.size)
2019-06-12 05:29:46 +00:00
else:
size = ''
2019-06-12 05:29:46 +00:00
2019-12-10 20:46:45 +00:00
bg = 'ddd' if shaded else 'fff'
anchor = atag(path, display_name=display_name)
2019-06-12 05:29:46 +00:00
row = form.format(
bg=bg,
2019-12-10 20:46:45 +00:00
anchor=anchor,
2019-06-12 05:29:46 +00:00
size=size,
)
return row
def path_to_url(path):
url = path.relative_path[2:]
url = url.replace(os.sep, '/')
url = '/' + url
url = urllib.parse.quote(url)
return url
def url_to_path(path):
path = urllib.parse.unquote(path)
path = path.strip('/')
return pathclass.Path(path)
def zip_directory(path):
zipfile = zipstream.ZipFile(mode='w', compression=zipstream.ZIP_STORED)
for item in path.walk():
if item.is_dir:
continue
arcname = item.relative_to(path).lstrip('.' + os.sep)
zipfile.write(filename=item.absolute_path, arcname=arcname)
return zipfile
2020-10-10 15:44:56 +00:00
# COMMAND LINE ###################################################################################################
2016-07-05 07:24:08 +00:00
2020-01-20 05:19:45 +00:00
def simpleserver_argparse(args):
2020-10-10 15:44:56 +00:00
server = SimpleServer(port=args.port, password=args.password)
server.start()
2020-01-20 05:19:45 +00:00
def main(argv):
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('port', nargs='?', type=int, default=40000)
parser.add_argument('--password', dest='password', default=None)
parser.set_defaults(func=simpleserver_argparse)
args = parser.parse_args(argv)
return args.func(args)
2016-07-20 03:31:47 +00:00
if __name__ == '__main__':
2020-01-20 05:19:45 +00:00
raise SystemExit(main(sys.argv[1:]))