decorated_function() - Code Metrics - Inspection of "Untested code / adding a password option" - ergoithz/browsepy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#21)

unknown

created 2017-06-30 12:06 UTC

decorated_function() A

↳ Parent: login_required()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	5
rs	9.4285
cc	2

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

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

from functools import wraps
from flask import g, request, redirect, url_for

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

from .manager import PluginManager
from .file import Node, OutsideRemovableBase, OutsideDirectoryBase, \
                  secure_filename
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)

# if a password is not provided in args consider the user logged
if app.config.password :
    g.user_is_logged = False
else :
    g.user_is_logged = True



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))

def login_required(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if g.user is None: # TODO implement if pass ...
            return redirect(url_for('login', next=request.url))
        return f(*args, **kwargs)
    return decorated_function

@app.route("/login", methods=["GET", "POST"])
def login():
    if not g.user_is_logged :
        if request.method == 'POST':
            password = request.form['password']
            if password == app.config.password:
                g.user_is_logged = True
                return redirect(request.args.get("next"))
            else:
                return abort(401)
        else:
            return Response('''
            <form action="" method="post">
                <p><input type=password name=password>
                <p><input type=submit value=Login>
            </form>
            ''')
    else :
        return redirect(request.args.get("next"))


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

ergoithz / browsepy

Pull Request — master (#21)

decorated_function() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like