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
|
# SPDX-License-Identifier: GPL-3.0-or-later
# Proxy web UI packages loading.
#
# This file is part of Hydrilla&Haketilo.
#
# Copyright (C) 2022 Wojtek Kosior
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
#
#
# I, Wojtek Kosior, thereby promise not to sue for violation of this
# file's license. Although I request that you do not make use this code
# in a proprietary program, I am not going to enforce this in court.
"""
.....
"""
# Enable using with Python 3.7.
from __future__ import annotations
import tempfile
import zipfile
import typing as t
from urllib.parse import urlparse
from pathlib import Path
import flask
import werkzeug
from ...exceptions import HaketiloException
from ...translations import smart_gettext as _
from .. import state as st
from . import _app
class InvalidUploadedMalcontent(HaketiloException):
def __init__(self):
super().__init__(_('err.proxy.uploaded_malcontent_invalid'))
bp = flask.Blueprint('load_packages', __package__)
@bp.route('/packages/load_from_disk', methods=['GET'])
def load_from_disk_get() -> flask.Response:
html = flask.render_template('packages__load_from_disk.html.jinja')
return flask.make_response(html, 200)
@bp.route('/packages/load_from_disk', methods=['POST'])
def load_from_disk_post() -> werkzeug.Response:
parsed_url = urlparse(flask.request.referrer)
if parsed_url.netloc != 'hkt.mitm.it':
return load_from_disk_get()
zip_file_storage = flask.request.files.get('packages_zipfile')
if zip_file_storage is None:
return load_from_disk_get()
with tempfile.TemporaryDirectory() as tmpdir_str:
tmpdir = Path(tmpdir_str)
tmpdir_child = tmpdir / 'childdir'
tmpdir_child.mkdir()
try:
with zipfile.ZipFile(zip_file_storage) as zip_file:
zip_file.extractall(tmpdir_child)
except:
raise HaketiloException(_('err.proxy.uploaded_file_not_zip'))
extracted_top_level_files = tuple(tmpdir_child.iterdir())
if extracted_top_level_files == ():
raise InvalidUploadedMalcontent()
if len(extracted_top_level_files) == 1 and \
extracted_top_level_files[0].is_dir():
malcontent_dir_path = extracted_top_level_files[0]
else:
malcontent_dir_path = tmpdir_child
state = t.cast(_app.WebUIApp, flask.current_app)._haketilo_state
try:
state.import_packages(malcontent_dir_path)
except:
raise InvalidUploadedMalcontent()
return flask.redirect(flask.url_for('.packages'))
@bp.route('/packages')
def packages() -> flask.Response:
state = t.cast(_app.WebUIApp, flask.current_app)._haketilo_state
display_infos = state.mapping_version_store().get_display_infos()
sorted_infos = sorted(display_infos, key=(lambda di: di.info))
html = flask.render_template(
'packages.html.jinja',
display_infos = sorted_infos
)
return flask.make_response(html, 200)
@bp.route('/packages/view/<string:mapping_id>')
def show_package(mapping_id: str) -> flask.Response:
state = t.cast(_app.WebUIApp, flask.current_app)._haketilo_state
try:
store = state.mapping_version_store()
display_info = store.get(mapping_id).get_display_info()
html = flask.render_template(
'packages__show_single.html.jinja',
display_info = display_info
)
return flask.make_response(html, 200)
except st.MissingItemError:
flask.abort(404)
|
