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
|
import http.server
import logging
import os
import sqlite3
import tempfile
import db
import templates
import utils
logger = logging.getLogger(__name__)
conn = sqlite3.connect('db.sqlite3')
files_directory = 'files'
authorized_key = os.environ['KEY']
class MyServer(http.server.BaseHTTPRequestHandler):
def do_GET(self):
match self.path:
case '/':
self._serve_str(templates.index, 200, 'text/html')
case '/main.js':
self._serve_file('public/main.js', 'application/javascript')
case '/main.css':
self._serve_file('public/main.css', 'text/css')
case path:
if path.endswith('?download'):
download = True
path = path[:-len('?download')]
else:
download = False
file_id = path[1:]
res = db.get_file(conn, file_id)
if res is None:
self._serve_str(templates.not_found, 404, 'text/html')
else:
filename, expires, content_length = res
disk_path = os.path.join(files_directory, file_id)
if download:
headers = [
('Content-Disposition', f'attachment; filename={filename}'),
('Content-Length', content_length)
]
self._serve_file(disk_path, 'application/octet-stream', headers)
else:
href = f'{file_id}?download'
self._serve_str(templates.download(href, filename, expires), 200, 'text/html')
def do_POST(self):
key = self.headers['X-Key']
if not key == authorized_key:
logging.info('Unauthorized to upload file: wrong key')
self._serve_str('Unauthorized', 401)
else:
logging.info('Uploading file')
content_length = int(self.headers['content-length'])
filename = utils.sanitize_filename(self.headers['X-FileName'])
expiration = self.headers['X-Expiration']
with tempfile.NamedTemporaryFile(delete = False) as tmp:
utils.transfer(self.rfile, tmp, content_length = content_length)
logging.info('File uploaded')
file_id = db.insert_file(conn, filename, expiration, content_length)
os.makedirs(files_directory, exist_ok=True)
os.rename(tmp.name, os.path.join(files_directory, file_id))
self._serve_str(file_id, 200)
def _serve_str(self, s, code, content_type='text/plain'):
self.send_response(code)
self.send_header('Content-type', content_type)
self.end_headers()
self.wfile.write(bytes(s, 'utf-8'))
def _serve_file(self, filename, content_type, headers = []):
self.send_response(200)
self.send_header('Content-type', content_type)
for header_name, header_value in headers:
self.send_header(header_name, header_value)
self.end_headers()
with open(filename, 'rb') as f:
utils.transfer(f, self.wfile)
|
