bad_request_error() - Code Metrics - Inspection of "add and handle upload exceptions" - ergoithz/browsepy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — pathconf ( 6b4a1f...acbd18 )

by Felipe A.

created 2017-09-08 12:42 UTC

bad_request_error() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	4
rs	10
cc	2

#!/usr/bin/env python
# -*- coding: UTF-8 -*-

import logging
import os
import os.path
import json
import base64

from flask import Response, request, render_template, redirect, \
                  url_for, send_from_directory, stream_with_context, \
                  make_response
from werkzeug.exceptions import NotFound

from .appconfig import Flask
from .manager import PluginManager
from .file import Node, secure_filename
from .exceptions import OutsideRemovableBase, OutsideDirectoryBase, \
                        InvalidFilenameError, InvalidPathError
from . import compat
from . import __meta__ as meta

__app__ = meta.app  # noqa
__version__ = meta.version  # noqa
__license__ = meta.license  # noqa
__author__ = meta.author  # noqa
__basedir__ = os.path.abspath(os.path.dirname(compat.fsdecode(__file__)))

logger = logging.getLogger(__name__)

app = Flask(
    __name__,
    static_url_path='/static',
    static_folder=os.path.join(__basedir__, "static"),
    template_folder=os.path.join(__basedir__, "templates")
    )
app.config.update(
    directory_base=compat.getcwd(),
    directory_start=compat.getcwd(),
    directory_remove=None,
    directory_upload=None,
    directory_tar_buffsize=262144,
    directory_downloadable=True,
    use_binary_multiples=True,
    plugin_modules=[],
    plugin_namespaces=(
        'browsepy.plugin',
        'browsepy_',
        '',
        ),
    exclude_fnc=None,
    )
app.jinja_env.add_extension('browsepy.transform.htmlcompress.HTMLCompress')

if 'BROWSEPY_SETTINGS' in os.environ:
    app.config.from_envvar('BROWSEPY_SETTINGS')

plugin_manager = PluginManager(app)


def iter_cookie_browse_sorting(cookies):
    '''
    Get sorting-cookie from cookies dictionary.

    :yields: tuple of path and sorting property
    :ytype: 2-tuple of strings
    '''
    try:
        data = cookies.get('browse-sorting', 'e30=').encode('ascii')
        for path, prop in json.loads(base64.b64decode(data).decode('utf-8')):
            yield path, prop
    except (ValueError, TypeError, KeyError) as e:
        logger.exception(e)


def get_cookie_browse_sorting(path, default):
    '''
    Get sorting-cookie data for path of current request.

    :returns: sorting property
    :rtype: string
    '''
    if request:
        for cpath, cprop in iter_cookie_browse_sorting(request.cookies):
            if path == cpath:
                return cprop
    return default


def browse_sortkey_reverse(prop):
    '''
    Get sorting function for directory listing based on given attribute
    name, with some caveats:
    * Directories will be first.
    * If *name* is given, link widget lowercase text will be used istead.
    * If *size* is given, bytesize will be used.

    :param prop: file attribute name
    :returns: tuple with sorting gunction and reverse bool
    :rtype: tuple of a dict and a bool
    '''
    if prop.startswith('-'):
        prop = prop[1:]
        reverse = True
    else:
        reverse = False

    if prop == 'text':
        return (
            lambda x: (
                x.is_directory == reverse,
                x.link.text.lower() if x.link and x.link.text else x.name
                ),
            reverse
            )
    if prop == 'size':
        return (
            lambda x: (
                x.is_directory == reverse,
                x.stats.st_size
                ),
            reverse
            )
    return (
        lambda x: (
            x.is_directory == reverse,
            getattr(x, prop, None)
            ),
        reverse
        )


def stream_template(template_name, **context):
    '''
    Some templates can be huge, this function returns an streaming response,
    sending the content in chunks and preventing from timeout.

    :param template_name: template
    :param **context: parameters for templates.
    :yields: HTML strings
    '''
    app.update_template_context(context)
    template = app.jinja_env.get_template(template_name)
    stream = template.generate(context)
    return Response(stream_with_context(stream))


@app.context_processor
def template_globals():
    return {
        'manager': app.extensions['plugin_manager'],
        'len': len,
        }


@app.route('/sort/<string:property>', defaults={"path": ""})
@app.route('/sort/<string:property>/<path:path>')
def sort(property, path):
    try:
        directory = Node.from_urlpath(path)
    except OutsideDirectoryBase:
        return NotFound()

    if not directory.is_directory or directory.is_excluded:
        return NotFound()

    data = [
        (cpath, cprop)
        for cpath, cprop in iter_cookie_browse_sorting(request.cookies)
        if cpath != path
        ]
    data.append((path, property))
    raw_data = base64.b64encode(json.dumps(data).encode('utf-8'))

    # prevent cookie becoming too large
    while len(raw_data) > 3975:  # 4000 - len('browse-sorting=""; Path=/')
        data.pop(0)
        raw_data = base64.b64encode(json.dumps(data).encode('utf-8'))

    response = redirect(url_for(".browse", path=directory.urlpath))
    response.set_cookie('browse-sorting', raw_data)
    return response


@app.route("/browse", defaults={"path": ""})
@app.route('/browse/<path:path>')
def browse(path):
    sort_property = get_cookie_browse_sorting(path, 'text')
    sort_fnc, sort_reverse = browse_sortkey_reverse(sort_property)

    try:
        directory = Node.from_urlpath(path)
        if directory.is_directory and not directory.is_excluded:
            return stream_template(
                'browse.html',
                file=directory,
                sort_property=sort_property,
                sort_fnc=sort_fnc,
                sort_reverse=sort_reverse
                )
    except OutsideDirectoryBase:
        pass
    return NotFound()


@app.route('/open/<path:path>', endpoint="open")
def open_file(path):
    try:
        file = Node.from_urlpath(path)
        if file.is_file and not file.is_excluded:
            return send_from_directory(file.parent.path, file.name)
    except OutsideDirectoryBase:
        pass
    return NotFound()


@app.route("/download/file/<path:path>")
def download_file(path):
    try:
        file = Node.from_urlpath(path)
        if file.is_file and not file.is_excluded:
            return file.download()
    except OutsideDirectoryBase:
        pass
    return NotFound()


@app.route("/download/directory/<path:path>.tgz")
def download_directory(path):
    try:
        directory = Node.from_urlpath(path)
        if directory.is_directory and not directory.is_excluded:
            return directory.download()
    except OutsideDirectoryBase:
        pass
    return NotFound()


@app.route("/remove/<path:path>", methods=("GET", "POST"))
def remove(path):
    try:
        file = Node.from_urlpath(path)
    except OutsideDirectoryBase:
        return NotFound()

    if not file.can_remove or file.is_excluded:
        return NotFound()

    if request.method == 'GET':
        return render_template('remove.html', file=file)

    parent = file.parent
    if parent is None:
        # base is not removable
        return NotFound()

    try:
        file.remove()
    except OutsideRemovableBase:
        return NotFound()

    return redirect(url_for(".browse", path=parent.urlpath))


@app.route("/upload", defaults={'path': ''}, methods=("POST",))
@app.route("/upload/<path:path>", methods=("POST",))
def upload(path):
    try:
        directory = Node.from_urlpath(path)
    except OutsideDirectoryBase:
        return NotFound()

    if (
      not directory.is_directory or
      not directory.can_upload or
      directory.is_excluded
      ):
        return NotFound()

    for v in request.files.listvalues():
        for f in v:
            filename = secure_filename(f.filename)
            if filename:
                filename = directory.choose_filename(filename)
                filepath = os.path.join(directory.path, filename)
                f.save(filepath)
            else:
                raise InvalidFilenameError(filename=f.filename)
    return redirect(url_for(".browse", path=directory.urlpath))


@app.route("/")
def index():
    path = app.config["directory_start"] or app.config["directory_base"]
    try:
        urlpath = Node(path).urlpath
    except OutsideDirectoryBase:
        return NotFound()
    return browse(urlpath)


@app.after_request
def page_not_found(response):
    if response.status_code == 404:
        return make_response((render_template('404.html'), 404))
    return response


@app.errorhandler(InvalidPathError)
def bad_request_error(e):
    file = Node(e.path) if isinstance(e, InvalidPathError) else None
    return render_template('error.upload.html', file=file, error=e), 400


@app.errorhandler(404)
def page_not_found_error(e):
    return render_template('404.html'), 404


@app.errorhandler(500)
def internal_server_error(e):  # pragma: no cover
    logger.exception(e)
    return getattr(e, 'message', 'Internal server error'), 500


1			#!/usr/bin/env python
2			# -- coding: UTF-8 --
3
4			import logging
5			import os
6			import os.path
7			import json
8			import base64
9
10			from flask import Response, request, render_template, redirect, \
11			url_for, send_from_directory, stream_with_context, \
12			make_response
13			from werkzeug.exceptions import NotFound
14
15			from .appconfig import Flask
16			from .manager import PluginManager
17			from .file import Node, secure_filename
18			from .exceptions import OutsideRemovableBase, OutsideDirectoryBase, \
19			InvalidFilenameError, InvalidPathError
20			from . import compat
21			from . import __meta__ as meta
22
23			__app__ = meta.app # noqa
24			__version__ = meta.version # noqa
25			__license__ = meta.license # noqa
26			__author__ = meta.author # noqa
27			__basedir__ = os.path.abspath(os.path.dirname(compat.fsdecode(__file__)))
28
29			logger = logging.getLogger(__name__)
30
31			app = Flask(
32			__name__,
33			static_url_path='/static',
34			static_folder=os.path.join(__basedir__, "static"),
35			template_folder=os.path.join(__basedir__, "templates")
36			)
37			app.config.update(
38			directory_base=compat.getcwd(),
39			directory_start=compat.getcwd(),
40			directory_remove=None,
41			directory_upload=None,
42			directory_tar_buffsize=262144,
43			directory_downloadable=True,
44			use_binary_multiples=True,
45			plugin_modules=[],
46			plugin_namespaces=(
47			'browsepy.plugin',
48			'browsepy_',
49			'',
50			),
51			exclude_fnc=None,
52			)
53			app.jinja_env.add_extension('browsepy.transform.htmlcompress.HTMLCompress')
54
55			if 'BROWSEPY_SETTINGS' in os.environ:
56			app.config.from_envvar('BROWSEPY_SETTINGS')
57
58			plugin_manager = PluginManager(app)
59
60
61			def iter_cookie_browse_sorting(cookies):
62			'''
63			Get sorting-cookie from cookies dictionary.
64
65			:yields: tuple of path and sorting property
66			:ytype: 2-tuple of strings
67			'''
68			try:
69			data = cookies.get('browse-sorting', 'e30=').encode('ascii')
70			for path, prop in json.loads(base64.b64decode(data).decode('utf-8')):
71			yield path, prop
72			except (ValueError, TypeError, KeyError) as e:
73			logger.exception(e)
74
75
76			def get_cookie_browse_sorting(path, default):
77			'''
78			Get sorting-cookie data for path of current request.
79
80			:returns: sorting property
81			:rtype: string
82			'''
83			if request:
84			for cpath, cprop in iter_cookie_browse_sorting(request.cookies):
85			if path == cpath:
86			return cprop
87			return default
88
89
90			def browse_sortkey_reverse(prop):
91			'''
92			Get sorting function for directory listing based on given attribute
93			name, with some caveats:
94			* Directories will be first.
95			* If name is given, link widget lowercase text will be used istead.
96			* If size is given, bytesize will be used.
97
98			:param prop: file attribute name
99			:returns: tuple with sorting gunction and reverse bool
100			:rtype: tuple of a dict and a bool
101			'''
102			if prop.startswith('-'):
103			prop = prop[1:]
104			reverse = True
105			else:
106			reverse = False
107
108			if prop == 'text':
109			return (
110			lambda x: (
111			x.is_directory == reverse,
112			x.link.text.lower() if x.link and x.link.text else x.name
113			),
114			reverse
115			)
116			if prop == 'size':
117			return (
118			lambda x: (
119			x.is_directory == reverse,
120			x.stats.st_size
121			),
122			reverse
123			)
124			return (
125			lambda x: (
126			x.is_directory == reverse,
127			getattr(x, prop, None)
128			),
129			reverse
130			)
131
132
133			def stream_template(template_name, **context):
134			'''
135			Some templates can be huge, this function returns an streaming response,
136			sending the content in chunks and preventing from timeout.
137
138			:param template_name: template
139			:param **context: parameters for templates.
140			:yields: HTML strings
141			'''
142			app.update_template_context(context)
143			template = app.jinja_env.get_template(template_name)
144			stream = template.generate(context)
145			return Response(stream_with_context(stream))
146
147
148			@app.context_processor
149			def template_globals():
150			return {
151			'manager': app.extensions['plugin_manager'],
152			'len': len,
153			}
154
155
156			@app.route('/sort/<string:property>', defaults={"path": ""})
157			@app.route('/sort/<string:property>/<path:path>')
158			def sort(property, path):
159			try:
160			directory = Node.from_urlpath(path)
161			except OutsideDirectoryBase:
162			return NotFound()
163
164			if not directory.is_directory or directory.is_excluded:
165			return NotFound()
166
167			data = [
168			(cpath, cprop)
169			for cpath, cprop in iter_cookie_browse_sorting(request.cookies)
170			if cpath != path
171			]
172			data.append((path, property))
173			raw_data = base64.b64encode(json.dumps(data).encode('utf-8'))
174
175			# prevent cookie becoming too large
176			while len(raw_data) > 3975: # 4000 - len('browse-sorting=""; Path=/')
177			data.pop(0)
178			raw_data = base64.b64encode(json.dumps(data).encode('utf-8'))
179
180			response = redirect(url_for(".browse", path=directory.urlpath))
181			response.set_cookie('browse-sorting', raw_data)
182			return response
183
184
185			@app.route("/browse", defaults={"path": ""})
186			@app.route('/browse/<path:path>')
187			def browse(path):
188			sort_property = get_cookie_browse_sorting(path, 'text')
189			sort_fnc, sort_reverse = browse_sortkey_reverse(sort_property)
190
191			try:
192			directory = Node.from_urlpath(path)
193			if directory.is_directory and not directory.is_excluded:
194			return stream_template(
195			'browse.html',
196			file=directory,
197			sort_property=sort_property,
198			sort_fnc=sort_fnc,
199			sort_reverse=sort_reverse
200			)
201			except OutsideDirectoryBase:
202			pass
203			return NotFound()
204
205
206			@app.route('/open/<path:path>', endpoint="open")
207			def open_file(path):
208			try:
209			file = Node.from_urlpath(path)
210			if file.is_file and not file.is_excluded:
211			return send_from_directory(file.parent.path, file.name)
212			except OutsideDirectoryBase:
213			pass
214			return NotFound()
215
216
217			@app.route("/download/file/<path:path>")
218			def download_file(path):
219			try:
220			file = Node.from_urlpath(path)
221			if file.is_file and not file.is_excluded:
222			return file.download()
223			except OutsideDirectoryBase:
224			pass
225			return NotFound()
226
227
228			@app.route("/download/directory/<path:path>.tgz")
229			def download_directory(path):
230			try:
231			directory = Node.from_urlpath(path)
232			if directory.is_directory and not directory.is_excluded:
233			return directory.download()
234			except OutsideDirectoryBase:
235			pass
236			return NotFound()
237
238
239			@app.route("/remove/<path:path>", methods=("GET", "POST"))
240			def remove(path):
241			try:
242			file = Node.from_urlpath(path)
243			except OutsideDirectoryBase:
244			return NotFound()
245
246			if not file.can_remove or file.is_excluded:
247			return NotFound()
248
249			if request.method == 'GET':
250			return render_template('remove.html', file=file)
251
252			parent = file.parent
253			if parent is None:
254			# base is not removable
255			return NotFound()
256
257			try:
258			file.remove()
259			except OutsideRemovableBase:
260			return NotFound()
261
262			return redirect(url_for(".browse", path=parent.urlpath))
263
264
265			@app.route("/upload", defaults={'path': ''}, methods=("POST",))
266			@app.route("/upload/<path:path>", methods=("POST",))
267			def upload(path):
268			try:
269			directory = Node.from_urlpath(path)
270			except OutsideDirectoryBase:
271			return NotFound()
272
273			if (
274			not directory.is_directory or
275			not directory.can_upload or
276			directory.is_excluded
277			):
278			return NotFound()
279
280			for v in request.files.listvalues():
281			for f in v:
282			filename = secure_filename(f.filename)
283			if filename:
284			filename = directory.choose_filename(filename)
285			filepath = os.path.join(directory.path, filename)
286			f.save(filepath)
287			else:
288			raise InvalidFilenameError(filename=f.filename)
289			return redirect(url_for(".browse", path=directory.urlpath))
290
291
292			@app.route("/")
293			def index():
294			path = app.config["directory_start"] or app.config["directory_base"]
295			try:
296			urlpath = Node(path).urlpath
297			except OutsideDirectoryBase:
298			return NotFound()
299			return browse(urlpath)
300
301
302			@app.after_request
303			def page_not_found(response):
304			if response.status_code == 404:
305			return make_response((render_template('404.html'), 404))
306			return response
307
308
309			@app.errorhandler(InvalidPathError)
310			def bad_request_error(e):
311			file = Node(e.path) if isinstance(e, InvalidPathError) else None
312			return render_template('error.upload.html', file=file, error=e), 400
313
314
315			@app.errorhandler(404)
316			def page_not_found_error(e):
317			return render_template('404.html'), 404
318
319
320			@app.errorhandler(500)
321			def internal_server_error(e): # pragma: no cover
322			logger.exception(e)
323			return getattr(e, 'message', 'Internal server error'), 500
324

ergoithz / browsepy

Push — pathconf ( 6b4a1f...acbd18 )

bad_request_error() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like