Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 26 additions & 10 deletions flask_uploader.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,12 +14,28 @@
with open(os.path.join(current_path, 'UploaderConfig.json'), 'r') as f:
config = json.load(f)

def api_check_admin(token: str):
def auth_headers_from_request():
headers = {}
token = flask.request.headers.get('Bearer', None)
if token:
headers["Bearer"] = token
session_id = flask.request.cookies.get('AuthSession', None)
if session_id:
headers["Cookie"] = f"AuthSession={session_id}"
return headers


def api_check_admin(auth_headers: dict):
if not config["check_admin"]:
return True
if not token:
if not auth_headers:
return False
r = requests.get("https://letovocorp.ru/api/auth/amiadmin", headers={"Bearer": token}, verify=False)
r = requests.get(
"https://letovocorp.ru/letovo-api/auth/amiuploader",
headers=auth_headers,
verify=False,
timeout=10,
)
return json.loads(r.text)["status"] == 't'

@app.route('/', methods=['POST'])
Expand All @@ -36,9 +52,9 @@ def upload_file():
file_path = os.path.join(ROOT_PATH, config["paths"][config["supported"][extention]])
else:
file_path = os.path.join(ROOT_PATH, config["paths"]["other"])
token = flask.request.headers.get('Bearer', None)
if not api_check_admin(token):
return f"You are not admin, your token: {token}", 403
auth_headers = auth_headers_from_request()
if not api_check_admin(auth_headers):
return "Not authorized", 403

file.save(os.path.join(file_path, file.filename))

Expand All @@ -52,13 +68,13 @@ def upload_avatar():
if file.filename == '':
return "No selected file", 400
file_path = os.path.join(ROOT_PATH, config["ava_path"])
token = flask.request.headers.get('Bearer', None)
if not api_check_admin(token):
return f"You are not admin, your token: {token}", 403
auth_headers = auth_headers_from_request()
if not api_check_admin(auth_headers):
return "Not authorized", 403
file.filename.replace(" ", "_")
file.filename = str(datetime.now().timestamp()).replace('.', '_') + "_" + file.filename
file.save(os.path.join(file_path, file.filename))
return '{"file": "/' + str(os.path.join(config["ava_path"], file.filename)) + '"}'

if __name__ == '__main__':
app.run(host='0.0.0.0', port=8880, debug=True, threaded=True, use_reloader=False)
app.run(host='0.0.0.0', port=8880, debug=False, threaded=True, use_reloader=False)
82 changes: 82 additions & 0 deletions test_flask_uploader_auth.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
import io
import json

import flask_uploader


def _configure_tmp(monkeypatch, tmp_path):
monkeypatch.setattr(flask_uploader, "ROOT_PATH", str(tmp_path))
monkeypatch.setitem(flask_uploader.config, "check_admin", True)
monkeypatch.setitem(flask_uploader.config, "supported", {"txt": "other"})
monkeypatch.setitem(flask_uploader.config, "paths", {"other": "other"})
monkeypatch.setitem(flask_uploader.config, "ava_path", "avatars")
(tmp_path / "other").mkdir()
(tmp_path / "avatars").mkdir()


def test_upload_accepts_auth_session_cookie(monkeypatch, tmp_path):
_configure_tmp(monkeypatch, tmp_path)
seen_headers = {}

class Response:
text = json.dumps({"status": "t"})

def fake_get(url, headers=None, **kwargs):
seen_headers.update(headers or {})
return Response()

monkeypatch.setattr(flask_uploader.requests, "get", fake_get)
client = flask_uploader.app.test_client()
client.set_cookie("AuthSession", "session-123")

response = client.post(
"/",
data={"file": (io.BytesIO(b"hello"), "hello.txt")},
content_type="multipart/form-data",
)

assert response.status_code == 200
assert seen_headers == {"Cookie": "AuthSession=session-123"}


def test_upload_keeps_bearer_header_auth(monkeypatch, tmp_path):
_configure_tmp(monkeypatch, tmp_path)
seen_headers = {}

class Response:
text = json.dumps({"status": "t"})

def fake_get(url, headers=None, **kwargs):
seen_headers.update(headers or {})
return Response()

monkeypatch.setattr(flask_uploader.requests, "get", fake_get)
client = flask_uploader.app.test_client()

response = client.post(
"/",
data={"file": (io.BytesIO(b"hello"), "hello.txt")},
headers={"Bearer": "legacy-token"},
content_type="multipart/form-data",
)

assert response.status_code == 200
assert seen_headers == {"Bearer": "legacy-token"}


def test_upload_without_auth_is_rejected_before_backend_call(monkeypatch, tmp_path):
_configure_tmp(monkeypatch, tmp_path)

def fail_get(*args, **kwargs):
raise AssertionError("auth backend should not be called without credentials")

monkeypatch.setattr(flask_uploader.requests, "get", fail_get)
client = flask_uploader.app.test_client()

response = client.post(
"/",
data={"file": (io.BytesIO(b"hello"), "hello.txt")},
content_type="multipart/form-data",
)

assert response.status_code == 403
Loading