summaryrefslogtreecommitdiff
path: root/fbin
diff options
context:
space:
mode:
authorJon Bergli Heier <snakebite@jvnv.net>2021-06-09 19:19:56 +0200
committerJon Bergli Heier <snakebite@jvnv.net>2021-06-09 19:21:34 +0200
commite96bedf7477d392b8821f76ca85038c198c84375 (patch)
treec56deba384f45da8958d848113b4da9fa73fe3f9 /fbin
parent7c13b1038482d68c2ab581dad16fa81c0a09034e (diff)
Fix linting errors
Style, unused imports, unused variables, etc. as reported by flake8. Configuration for flake8 has been added to setup.cfg.
Diffstat (limited to 'fbin')
-rw-r--r--fbin/__init__.py10
-rw-r--r--fbin/api.py20
-rw-r--r--fbin/db.py14
-rwxr-xr-xfbin/fbin.py92
-rw-r--r--fbin/file_storage/base.py3
-rw-r--r--fbin/file_storage/exceptions.py8
-rw-r--r--fbin/file_storage/filesystem.py7
-rw-r--r--fbin/file_storage/s3.py7
-rw-r--r--fbin/login.py25
-rw-r--r--fbin/monkey.py6
10 files changed, 120 insertions, 72 deletions
diff --git a/fbin/__init__.py b/fbin/__init__.py
index 4bc2ef3..f36d74b 100644
--- a/fbin/__init__.py
+++ b/fbin/__init__.py
@@ -7,14 +7,14 @@ app.config.from_pyfile('fbin.cfg')
# Set up some custom converters. These are needed for file URLs to be properly parsed.
+
class HashConverter(BaseConverter):
regex = r'\w+'
+
class ExtensionConverter(BaseConverter):
regex = r'\.\w+'
-app.url_map.converters['hash'] = HashConverter
-app.url_map.converters['ext'] = ExtensionConverter
@app.context_processor
def context_processors():
@@ -29,11 +29,15 @@ def context_processors():
'nav_html': nav_html,
}
+
+app.url_map.converters['hash'] = HashConverter
+app.url_map.converters['ext'] = ExtensionConverter
+
with app.app_context():
from .fbin import app as fbin, db
from .api import app as api
from .login import login_manager
app.register_blueprint(fbin)
- app.register_blueprint(api, url_prefix = '/api')
+ app.register_blueprint(api, url_prefix='/api')
login_manager.init_app(app)
db.init_app(app)
diff --git a/fbin/api.py b/fbin/api.py
index 8f3f86c..5659b3a 100644
--- a/fbin/api.py
+++ b/fbin/api.py
@@ -11,6 +11,7 @@ from .fbin import upload as fbin_upload, get_file
app = Blueprint('api', __name__)
+
def makejson(f):
@functools.wraps(f)
def wrapper(*args, **kwargs):
@@ -20,16 +21,17 @@ def makejson(f):
return r
return wrapper
+
@app.before_request
def authenticate():
g.user = None
- if not 'Authorization' in request.headers:
+ if 'Authorization' not in request.headers:
abort(403)
scheme, token = request.headers['Authorization'].split(None, 1)
if scheme != 'Bearer':
abort(400)
try:
- token = jwt.decode(token, current_app.config['SECRET_KEY'], issuer = request.url_root)
+ token = jwt.decode(token, current_app.config['SECRET_KEY'], issuer=request.url_root)
except jwt.InvalidTokenError:
abort(403)
try:
@@ -43,6 +45,7 @@ def authenticate():
except NoResultFound:
abort(403)
+
def api_login_required(f):
def wrapper(*args, **kwargs):
if not current_user.is_authenticated:
@@ -53,15 +56,17 @@ def api_login_required(f):
return f(*args, **kwargs)
return wrapper
-@app.route('/upload', methods = ['POST'])
+
+@app.route('/upload', methods=['POST'])
def upload():
- return fbin_upload(api = True, user = g.user)
+ return fbin_upload(api=True, user=g.user)
+
class FileAPI(MethodView):
decorators = [api_login_required, makejson]
def put(self, hash):
- f = get_file(hash, user_id = current_user.get_id())
+ f = get_file(hash, user_id=current_user.get_id())
if not f:
return {
'status': False,
@@ -83,8 +88,9 @@ class FileAPI(MethodView):
pass
# TODO: Add back FileAPI when ready.
-#file_api_view = FileAPI.as_view('file_api')
-#app.add_url_rule('/file/<hash>', view_func = file_api_view, methods = ['PUT', 'DELETE'])
+# file_api_view = FileAPI.as_view('file_api')
+# app.add_url_rule('/file/<hash>', view_func=file_api_view, methods=['PUT', 'DELETE'])
+
@app.route('/test')
def test():
diff --git a/fbin/db.py b/fbin/db.py
index 53b0d8b..9493f65 100644
--- a/fbin/db.py
+++ b/fbin/db.py
@@ -1,15 +1,15 @@
-from contextlib import contextmanager
import datetime
import mimetypes
import os
-from flask import current_app
-from sqlalchemy.orm.exc import NoResultFound
-from sqlalchemy.exc import IntegrityError
+# FIXME: reimported from elsewhere
+from sqlalchemy.orm.exc import NoResultFound # noqa: F401
+from sqlalchemy.exc import IntegrityError # noqa: F401
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
+
class User(db.Model):
__tablename__ = 'users'
@@ -23,6 +23,7 @@ class User(db.Model):
self.username = username
self.jab_id = jab_id
+
class UserSession(db.Model):
__tablename__ = 'sessions'
@@ -37,6 +38,7 @@ class UserSession(db.Model):
self.access_token = access_token
self.refresh_token = refresh_token
+
class File(db.Model):
__tablename__ = 'files'
@@ -51,7 +53,7 @@ class File(db.Model):
scanned = db.Column(db.Boolean, nullable=False, default=False)
blocked_reason = db.Column(db.JSON)
- def __init__(self, hash, filename, size, date, user_id = None, ip = None):
+ def __init__(self, hash, filename, size, date, user_id=None, ip=None):
self.hash = hash
self.filename = filename
self.size = size
@@ -80,7 +82,7 @@ class File(db.Model):
return self.date.strftime('%Y-%m-%d %H:%M:%S UTC')
def get_mime_type(self):
- return mimetypes.guess_type(self.filename, strict = False)[0] or 'application/octet-stream'
+ return mimetypes.guess_type(self.filename, strict=False)[0] or 'application/octet-stream'
def is_image(self):
return self.get_mime_type().startswith('image')
diff --git a/fbin/fbin.py b/fbin/fbin.py
index a195594..b062c9a 100755
--- a/fbin/fbin.py
+++ b/fbin/fbin.py
@@ -1,33 +1,28 @@
#!/usr/bin/env python
import base64
-import cgi
import datetime
-import hashlib
import importlib
-import io
-import json
-import mimetypes
import os
import random
import subprocess
import tempfile
-import urllib
from urllib.parse import urlencode, urljoin
-from flask import Blueprint, redirect, current_app, url_for, request, render_template, session, flash, send_file, abort, jsonify, Markup, Response
+from flask import Blueprint, redirect, current_app, url_for, request, render_template, session, \
+ flash, send_file, abort, jsonify, Response
from flask_login import login_user, logout_user, current_user, login_required
import jwt
-from PIL import Image, ExifTags
+from PIL import Image
import requests
-from werkzeug.utils import secure_filename
from .db import db, User, UserSession, File, NoResultFound, IntegrityError
from .monkey import patch as monkey_patch
from .login import login_manager, load_user
from .file_storage.exceptions import StorageError
-storage = importlib.import_module(current_app.config.get('STORAGE_MODULE', '.file_storage.filesystem'), package='fbin').Storage(current_app)
+storage = importlib.import_module(current_app.config.get('STORAGE_MODULE', '.file_storage.filesystem'), package='fbin') \
+ .Storage(current_app)
monkey_patch()
@@ -36,6 +31,7 @@ base62_alphabet = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXY
if not os.path.isdir(current_app.config['THUMB_DIRECTORY']):
os.mkdir(current_app.config['THUMB_DIRECTORY'])
+
def get_or_create_user(username, jab_id):
try:
return db.session.query(User).filter(User.jab_id == jab_id).one()
@@ -49,6 +45,7 @@ def get_or_create_user(username, jab_id):
except IntegrityError:
return None
+
def get_file(file_hash, user_id=None, update_accessed=False):
try:
f = db.session.query(File).filter(File.hash == file_hash)
@@ -65,6 +62,7 @@ def get_file(file_hash, user_id=None, update_accessed=False):
db.session.refresh(f)
return f
+
def get_files(user):
try:
db.session.add(user)
@@ -73,19 +71,23 @@ def get_files(user):
return []
return files
+
def delete_file(file):
db.session.delete(file)
db.session.commit()
storage.delete_file(file)
+
app = Blueprint('fbin', __name__)
+
@app.route('/')
def index():
return redirect(url_for('.upload'))
+
@app.route('/u')
-@app.route('/upload', methods = ['GET', 'POST'])
+@app.route('/upload', methods=['GET', 'POST'])
def upload(api=False, user=None):
def error(message):
if api:
@@ -131,23 +133,24 @@ def upload(api=False, user=None):
'status': True,
'hash': new_file.hash,
'urls': {
- 'base': url_for('fbin.file', hash = '', _external = True),
- 'full': url_for('fbin.file', hash = new_file.hash, filename = new_file.filename, _external = True),
- 'ext': url_for('fbin.file', hash = new_file.hash, ext = new_file.ext, _external = True),
- 'hash': url_for('fbin.file', hash = new_file.hash, _external = True),
+ 'base': url_for('fbin.file', hash='', _external=True),
+ 'full': url_for('fbin.file', hash=new_file.hash, filename=new_file.filename, _external=True),
+ 'ext': url_for('fbin.file', hash=new_file.hash, ext=new_file.ext, _external=True),
+ 'hash': url_for('fbin.file', hash=new_file.hash, _external=True),
},
})
elif old_api:
- return 'OK {hash}'.format(hash = new_file.hash)
+ return 'OK {hash}'.format(hash=new_file.hash)
else:
context = {
'file': new_file,
}
- return redirect(url_for('.uploaded', hash = new_file.hash))
+ return redirect(url_for('.uploaded', hash=new_file.hash))
+
@app.route('/uploaded/<hash>')
def uploaded(hash):
- f = get_file(hash, update_accessed = False)
+ f = get_file(hash, update_accessed=False)
if not f:
abort(404)
if f.user_id and (not current_user.is_authenticated or f.user_id != current_user.get_user_id()):
@@ -159,17 +162,18 @@ def uploaded(hash):
}
return render_template('uploaded.html', **context)
+
@app.route('/f/<hash:hash>')
@app.route('/f/<hash:hash><ext:ext>')
@app.route('/f/<hash:hash>/<path:filename>')
-@app.route('/file/<hash:hash>', endpoint = 'file')
-@app.route('/file/<hash:hash><ext:ext>', endpoint = 'file')
-@app.route('/file/<hash:hash>/<path:filename>', endpoint = 'file')
+@app.route('/file/<hash:hash>', endpoint='file')
+@app.route('/file/<hash:hash><ext:ext>', endpoint='file')
+@app.route('/file/<hash:hash>/<path:filename>', endpoint='file')
def _file(hash, ext=None, filename=None):
f = get_file(hash)
- if not f or (f.blocked_reason and (f.blocked_reason['positives'] >= current_app.config['VIRUSTOTAL_MINIMUM_POSITIVES'] \
- or any(scan['detected'] and scan['result'] in current_app.config['VIRUSTOTAL_SINGULAR_MATCHES']
- for scan in f.blocked_reason['scans'].values()))):
+ if not f or (f.blocked_reason and (f.blocked_reason['positives'] >= current_app.config['VIRUSTOTAL_MINIMUM_POSITIVES']
+ or any(scan['detected'] and scan['result'] in current_app.config['VIRUSTOTAL_SINGULAR_MATCHES']
+ for scan in f.blocked_reason['scans'].values()))):
abort(404)
path = storage.get_file(f)
if isinstance(path, Response):
@@ -186,6 +190,7 @@ def _file(hash, ext=None, filename=None):
mimetype = 'application/octet-stream'
return send_file(path, mimetype=mimetype, attachment_filename=f.filename)
+
@app.route('/l')
@app.route('/login')
def login():
@@ -197,23 +202,23 @@ def login():
'state': session['oauth_state'],
}))
+
@app.route('/account')
def account():
return redirect(current_app.config['ACCOUNT_URL'])
+
@app.route('/o')
@app.route('/logout')
def logout():
if not current_user.is_authenticated:
return redirect(url_for('.index'))
session_id = int(current_user.get_id().split(':', 1)[-1])
- try:
- db.session.query(UserSession).filter_by(id = session_id).delete()
- except:
- raise
+ db.session.query(UserSession).filter_by(id=session_id).delete()
logout_user()
return redirect(url_for('.index'))
+
@app.route('/auth')
def auth():
if 'error' in request.args:
@@ -234,7 +239,7 @@ def auth():
flash('Missing OAuth code', 'error')
return redirect(url_for('.index'))
rs = requests.Session()
- response = rs.post(urljoin(current_app.config['OAUTH_URL'], 'token'), data = {
+ response = rs.post(urljoin(current_app.config['OAUTH_URL'], 'token'), data={
'grant_type': 'authorization_code',
'code': code,
'client_id': current_app.config['OAUTH_CLIENT_ID'],
@@ -247,12 +252,15 @@ def auth():
flash(msg, 'error')
return redirect(url_for('.index'))
try:
- access_data = jwt.decode(token['access_token'], key = current_app.config['JWT_PUBLIC_KEY'], audience = current_app.config['OAUTH_CLIENT_ID'])
- refresh_data = jwt.decode(token['refresh_token'], key = current_app.config['JWT_PUBLIC_KEY'], audience = current_app.config['OAUTH_CLIENT_ID'])
+ jwt.decode(token['access_token'], key=current_app.config['JWT_PUBLIC_KEY'],
+ audience=current_app.config['OAUTH_CLIENT_ID'])
+ jwt.decode(token['refresh_token'], key=current_app.config['JWT_PUBLIC_KEY'],
+ audience=current_app.config['OAUTH_CLIENT_ID'])
except jwt.InvalidTokenError as e:
flash('Failed to verify token: {!s}'.format(e), 'error')
return redirect(url_for('.index'))
- response = rs.get(urljoin(current_app.config['OAUTH_URL'], '/api/user'), headers = {'Authorization': 'Bearer {}'.format(token['access_token'])})
+ response = rs.get(urljoin(current_app.config['OAUTH_URL'], '/api/user'),
+ headers={'Authorization': 'Bearer {}'.format(token['access_token'])})
user = response.json()
user = get_or_create_user(user['username'], user['id'])
us = UserSession(user.id, token['access_token'], token['refresh_token'])
@@ -264,9 +272,10 @@ def auth():
if not user:
flash('Failed to retrieve user instance.', 'error')
else:
- login_user(user, remember = True)
+ login_user(user, remember=True)
return redirect(url_for('.index'))
+
@app.route('/m')
@app.route('/files')
@login_required
@@ -279,11 +288,12 @@ def files():
}
return render_template('files.html', **context)
-@app.route('/files', methods = ['POST'])
+
+@app.route('/files', methods=['POST'])
@login_required
def file_edit():
user_id = int(current_user.get_id().split(':')[0])
- f = get_file(request.form.get('hash'), user_id = user_id, update_accessed = False)
+ f = get_file(request.form.get('hash'), user_id=user_id, update_accessed=False)
if not f:
flash('File not found.', 'error')
return redirect(url_for('.files'))
@@ -295,7 +305,7 @@ def file_edit():
elif 'delete' in request.form:
try:
delete_file(f)
- except:
+ except Exception:
flash('Failed to delete file.', 'error')
else:
flash('File deleted.', 'success')
@@ -303,6 +313,7 @@ def file_edit():
flash('No action was performed.', 'warning')
return redirect(url_for('.files'))
+
@app.route('/i')
@app.route('/images')
@login_required
@@ -316,6 +327,7 @@ def images():
}
return render_template('images.html', **context)
+
@app.route('/v')
@app.route('/videos')
@login_required
@@ -329,13 +341,14 @@ def videos():
}
return render_template('images.html', **context)
+
@app.route('/t/<hash:hash>')
@app.route('/thumb/<hash:hash>')
def thumb(hash):
f = get_file(hash, update_accessed=False)
response = storage.get_thumbnail(f)
if not response:
- with tempfile.NamedTemporaryFile(suffix='.jpg') as ttf: # temporary thumb file
+ with tempfile.NamedTemporaryFile(suffix='.jpg') as ttf: # temporary thumb file
if f.is_image():
try:
with storage.temp_file(f) as tf:
@@ -373,11 +386,13 @@ def thumb(hash):
return response
return send_file(response, attachment_filename='thumb.jpg')
+
@app.route('/h')
@app.route('/help')
def help():
return redirect(url_for('.api'))
+
@app.route('/api')
def api():
context = {
@@ -386,6 +401,7 @@ def api():
}
return render_template('api.html', **context)
+
@app.route('/generate-api-key')
def generate_api_key():
if not current_user.is_authenticated:
@@ -401,6 +417,7 @@ def generate_api_key():
token = jwt.encode(data, current_app.config['SECRET_KEY'])
return token
+
@app.route('/invalidate-api-keys')
@login_required
def invalidate_api_keys():
@@ -411,4 +428,5 @@ def invalidate_api_keys():
flash('All API keys invalidated.', 'success')
return redirect(request.referrer)
+
login_manager.login_view = '.login'
diff --git a/fbin/file_storage/base.py b/fbin/file_storage/base.py
index aa2c510..40c1588 100644
--- a/fbin/file_storage/base.py
+++ b/fbin/file_storage/base.py
@@ -1,7 +1,8 @@
import datetime
from ..db import db, File
-from .exceptions import *
+from .exceptions import FileSizeError
+
class BaseStorage:
def __init__(self, app):
diff --git a/fbin/file_storage/exceptions.py b/fbin/file_storage/exceptions.py
index 140221b..e949f20 100644
--- a/fbin/file_storage/exceptions.py
+++ b/fbin/file_storage/exceptions.py
@@ -1,2 +1,6 @@
-class StorageError(Exception): pass
-class FileSizeError(StorageError): pass
+class StorageError(Exception):
+ pass
+
+
+class FileSizeError(StorageError):
+ pass
diff --git a/fbin/file_storage/filesystem.py b/fbin/file_storage/filesystem.py
index b132264..5b4f0a3 100644
--- a/fbin/file_storage/filesystem.py
+++ b/fbin/file_storage/filesystem.py
@@ -4,6 +4,7 @@ import tempfile
from .base import BaseStorage
+
class Storage(BaseStorage):
def __init__(self, app):
super().__init__(app)
@@ -49,7 +50,7 @@ class Storage(BaseStorage):
file_path, size = self.upload_file(uploaded_file, file_hash, user)
try:
return self.add_file(file_hash, uploaded_file.filename, size, user, ip)
- except:
+ except: # noqa: E722; we want to unlink and re-raise on all exceptions
if os.path.exists(file_path):
os.unlink(file_path)
raise
@@ -89,7 +90,7 @@ class Storage(BaseStorage):
def store_thumbnail(self, f, stream):
path = self.get_thumb_path(f)
with open(path, 'wb') as f:
- buf = stream.read(1024*10)
+ buf = stream.read(1024 * 10)
while buf:
f.write(buf)
- buf = stream.read(1024*10)
+ buf = stream.read(1024 * 10)
diff --git a/fbin/file_storage/s3.py b/fbin/file_storage/s3.py
index 71ee1e7..b372dd7 100644
--- a/fbin/file_storage/s3.py
+++ b/fbin/file_storage/s3.py
@@ -7,6 +7,7 @@ from flask import request, send_file
from .base import BaseStorage
+
class Storage(BaseStorage):
def __init__(self, app):
super().__init__(app)
@@ -35,7 +36,7 @@ class Storage(BaseStorage):
obj, size = self.upload_file(uploaded_file, file_hash, user)
try:
return self.add_file(file_hash, uploaded_file.filename, size, user, ip)
- except:
+ except: # noqa: E722; we want to delete and re-raise on all exceptions
obj.delete()
raise
@@ -44,7 +45,7 @@ class Storage(BaseStorage):
bucket = self.app.config['S3_BUCKET']
obj = self.client.Object(bucket, key)
try:
- meta = obj.load()
+ obj.load()
return True
except botocore.exceptions.ClientError as e:
if e.response['Error']['Code'] == '404':
@@ -96,4 +97,4 @@ class Storage(BaseStorage):
def store_thumbnail(self, f, stream):
bucket = self.client.Bucket(self.app.config['S3_THUMB_BUCKET'])
key = self.get_object_key(f, thumb=True)
- obj = bucket.upload_fileobj(Fileobj=stream, Key=key)
+ bucket.upload_fileobj(Fileobj=stream, Key=key)
diff --git a/fbin/login.py b/fbin/login.py
index b9602a8..00f969d 100644
--- a/fbin/login.py
+++ b/fbin/login.py
@@ -11,6 +11,7 @@ from .db import db, User, UserSession
login_manager = LoginManager()
+
class BinUser:
def __init__(self, user, user_session):
self.user = user
@@ -18,23 +19,26 @@ class BinUser:
self.token = None
def refresh_access_token(self):
- response = requests.post(urljoin(current_app.config['OAUTH_URL'], 'token'), data = {
+ response = requests.post(urljoin(current_app.config['OAUTH_URL'], 'token'), data={
'grant_type': 'refresh_token',
'client_id': current_app.config['OAUTH_CLIENT_ID'],
'client_secret': current_app.config['OAUTH_CLIENT_SECRET'],
'refresh_token': self.user_session.refresh_token,
})
if response.status_code != 200:
- flash('Failed to refresh authentication token (API call returned {} {})'.format(response.status_code, response.reason), 'error')
+ flash('Failed to refresh authentication token (API call returned {} {})'.format(response.status_code,
+ response.reason), 'error')
return
token = response.json()
if 'error' in token:
flash('Failed to refresh authentication token ({})'.format(token['error']), 'error')
return
try:
- access_data = jwt.decode(token['access_token'], key = current_app.config['JWT_PUBLIC_KEY'], audience = current_app.config['OAUTH_CLIENT_ID'])
- refresh_data = jwt.decode(token['refresh_token'], key = current_app.config['JWT_PUBLIC_KEY'], audience = current_app.config['OAUTH_CLIENT_ID'])
- except jwt.InvalidTokenError as e:
+ jwt.decode(token['access_token'], key=current_app.config['JWT_PUBLIC_KEY'],
+ audience=current_app.config['OAUTH_CLIENT_ID'])
+ jwt.decode(token['refresh_token'], key=current_app.config['JWT_PUBLIC_KEY'],
+ audience=current_app.config['OAUTH_CLIENT_ID'])
+ except jwt.InvalidTokenError:
traceback.print_exc()
flash('Failed to refresh authentication token (verification failed)', 'error')
return
@@ -52,12 +56,13 @@ class BinUser:
if self.token:
return True
try:
- self.token = jwt.decode(self.user_session.access_token, key = current_app.config['JWT_PUBLIC_KEY'], audience = current_app.config['OAUTH_CLIENT_ID'])
+ self.token = jwt.decode(self.user_session.access_token, key=current_app.config['JWT_PUBLIC_KEY'],
+ audience=current_app.config['OAUTH_CLIENT_ID'])
except jwt.ExpiredSignatureError:
try:
if not self.refresh_access_token():
return False
- except:
+ except Exception:
traceback.print_exc()
flash('Failed to refresh authentication token (unhandled error; contact an admin)', 'error')
return False
@@ -83,12 +88,14 @@ class BinUser:
def username(self):
return self.user.username
+
@login_manager.user_loader
def load_user(user_id):
user_id, session_id = map(int, user_id.split(':', 1))
try:
- user, user_session = db.session.query(User, UserSession).join(UserSession).filter(User.id == user_id, UserSession.id == session_id).one()
+ user, user_session = db.session.query(User, UserSession).join(UserSession) \
+ .filter(User.id == user_id, UserSession.id == session_id).one()
return BinUser(user, user_session)
- except:
+ except Exception:
traceback.print_exc()
return None
diff --git a/fbin/monkey.py b/fbin/monkey.py
index c6458ad..89df4f3 100644
--- a/fbin/monkey.py
+++ b/fbin/monkey.py
@@ -4,6 +4,7 @@ from flask import current_app
import werkzeug.formparser
import werkzeug.wrappers
+
def werkzeug_patch():
global werkzeug_orig_stream_factory
@@ -11,15 +12,18 @@ def werkzeug_patch():
def custom_stream_factory(total_content_length, filename, content_type, content_length=None):
if total_content_length > 1024 * 500:
- return tempfile.NamedTemporaryFile('wb+', prefix = 'upload_', dir = current_app.config['FILE_DIRECTORY'], delete = True)
+ return tempfile.NamedTemporaryFile('wb+', prefix='upload_',
+ dir=current_app.config['FILE_DIRECTORY'], delete=True)
return werkzeug_orig_stream_factory(total_content_length, filename, content_type, content_length)
werkzeug.formparser.default_stream_factory = custom_stream_factory
werkzeug.wrappers.default_stream_factory = custom_stream_factory
+
def werkzeug_reset():
werkzeug.formparser.default_stream_factory = werkzeug_orig_stream_factory
werkzeug.wrappers.default_stream_factory = werkzeug_orig_stream_factory
+
def patch():
werkzeug_patch()