19114618549

Committed 05 Nov 2025 07:59PM UTC coverage: 94.093% (+0.03%) from 94.065%

Build # 19114618549

Build Type

Pull #1069

github

Committed by

peterkle

Commit Message

add more tests

Pull Request Pull Request #1069: RDISCROWD-8392 upgrade to boto3

Run Details

214 of 223 new or added lines in 7 files covered. (95.96%)

152 existing lines in 8 files now uncovered.

17920 of 19045 relevant lines covered (94.09%)

0.94 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.88

/pybossa/view/attachment.py

# -*- coding: utf8 -*-
# This file is part of PYBOSSA.
#
# Copyright (C) 2025 Scifabric LTD.
#
# PYBOSSA is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# PYBOSSA 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with PYBOSSA.  If not, see <http://www.gnu.org/licenses/>.
from flask import current_app, abort
from flask import Blueprint, request, Response
from flask_login import current_user, login_required
from werkzeug.exceptions import Forbidden, BadRequest

from pybossa.cloud_store_api.s3 import s3_get_email_attachment
from pybossa.core import signer
from pybossa.util import admin_or_project_owner
from itsdangerous import BadSignature

blueprint = Blueprint('attachment', __name__)

TASK_SIGNATURE_MAX_SIZE = 128

@login_required
@blueprint.route('/<string:signature>/<string:path>', methods=['GET', 'POST'])
def download_attachment(signature, path):
    """download attachment from storage location"""

    from pybossa.core import project_repo

    try:
        size_signature = len(signature)
        if size_signature > TASK_SIGNATURE_MAX_SIZE:
            current_app.logger.exception(
                "Invalid task signature. Signature length exceeds max allowed length. signature %s, path %s",
                signature, path)
            raise BadRequest('Invalid signature')
        
        signed_payload = signer.loads(signature)
        project_id = signed_payload.get("project_id")
        user_email = signed_payload.get("user_email")
        current_app.logger.info("download attachment url signed info. project id %d, user email %s", project_id, user_email)

        # admins and project owners are authorized to download the attachment
        if project_id:
            project = project_repo.get(project_id)
            if not project:
                raise BadRequest(f"Invalid project id {project_id}")

            current_app.logger.info("project info: id %d, owner ids %s. current user info: id %d, admin %r, authenticated %r",
                                    project.id, str(project.owners_ids), current_user.id, current_user.admin, current_user.is_authenticated)
            admin_or_project_owner(current_user, project)

        # admins or subadmin users tagged in signature can download the attachment
        if user_email:
            if not (current_user.admin or (current_user.subadmin and current_user.email_addr == user_email)):
                raise Forbidden('Access denied')

        resp = s3_get_email_attachment(path)
        response = Response(resp["content"], mimetype=resp["type"], status=200)
        if resp["content"]:
            response.headers['Content-Disposition'] = f'attachment; filename={resp["name"]}'
    except BadSignature as ex:
        current_app.logger.exception("BadSignature. %s, signature %s, path %s",str(ex), signature, path)
        response = Response(f"An internal error has occurred.", mimetype="text/plain", status=400)
    except BadRequest as ex:
        current_app.logger.exception("%s, signature %s, path %s",str(ex), signature, path)
        response = Response(f"An internal error has occurred.", mimetype="text/plain", status=400)
    except Forbidden as ex:
        current_app.logger.exception("%s, signature %s, path %s",str(ex), signature, path)
        response = Response(f"An internal error has occurred.", mimetype="text/plain", status=403)
    except Exception as ex:
        current_app.logger.exception("%s, signature %s, path %s", str(ex), signature, path)
        response = Response(f"Failed loading requested url.", mimetype="text/plain", status=400)
    return response

1	# -- coding: utf8 --
2	# This file is part of PYBOSSA.
3	#
4	# Copyright (C) 2025 Scifabric LTD.
5	#
6	# PYBOSSA is free software: you can redistribute it and/or modify
7	# it under the terms of the GNU Affero General Public License as published by
8	# the Free Software Foundation, either version 3 of the License, or
9	# (at your option) any later version.
10	#
11	# PYBOSSA is distributed in the hope that it will be useful,
12	# but WITHOUT ANY WARRANTY; without even the implied warranty of
13	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14	# GNU Affero General Public License for more details.
15	#
16	# You should have received a copy of the GNU Affero General Public License
17	# along with PYBOSSA. If not, see <http://www.gnu.org/licenses/>.
18	from flask import current_app, abort	1✔
19	from flask import Blueprint, request, Response	1✔
20	from flask_login import current_user, login_required	1✔
21	from werkzeug.exceptions import Forbidden, BadRequest	1✔
22
23	from pybossa.cloud_store_api.s3 import s3_get_email_attachment	1✔
24	from pybossa.core import signer	1✔
25	from pybossa.util import admin_or_project_owner	1✔
26	from itsdangerous import BadSignature	1✔
27
28	blueprint = Blueprint('attachment', __name__)	1✔
29
30	TASK_SIGNATURE_MAX_SIZE = 128	1✔
31
32	@login_required	1✔
33	@blueprint.route('/<string:signature>/<string:path>', methods=['GET', 'POST'])	1✔
34	def download_attachment(signature, path):	1✔
35	"""download attachment from storage location"""
36
37	from pybossa.core import project_repo	1✔
38
39	try:	1✔
40	size_signature = len(signature)	1✔
41	if size_signature > TASK_SIGNATURE_MAX_SIZE:	1✔
42	current_app.logger.exception(	1✔
43	"Invalid task signature. Signature length exceeds max allowed length. signature %s, path %s",
44	signature, path)
45	raise BadRequest('Invalid signature')	1✔
46
47	signed_payload = signer.loads(signature)	1✔
48	project_id = signed_payload.get("project_id")	1✔
49	user_email = signed_payload.get("user_email")	1✔
50	current_app.logger.info("download attachment url signed info. project id %d, user email %s", project_id, user_email)	1✔
51
52	# admins and project owners are authorized to download the attachment
53	if project_id:	1✔
54	project = project_repo.get(project_id)	1✔
55	if not project:	1✔
56	raise BadRequest(f"Invalid project id {project_id}")	1✔
57
58	current_app.logger.info("project info: id %d, owner ids %s. current user info: id %d, admin %r, authenticated %r",	1✔
59	project.id, str(project.owners_ids), current_user.id, current_user.admin, current_user.is_authenticated)
60	admin_or_project_owner(current_user, project)	1✔
61
62	# admins or subadmin users tagged in signature can download the attachment
63	if user_email:	1✔
64	if not (current_user.admin or (current_user.subadmin and current_user.email_addr == user_email)):	1✔
65	raise Forbidden('Access denied')	1✔
66
67	resp = s3_get_email_attachment(path)	1✔
68	response = Response(resp["content"], mimetype=resp["type"], status=200)	1✔
69	if resp["content"]:	1✔
70	response.headers['Content-Disposition'] = f'attachment; filename={resp["name"]}'	1✔
71	except BadSignature as ex:	1✔
72	current_app.logger.exception("BadSignature. %s, signature %s, path %s",str(ex), signature, path)	1✔
73	response = Response(f"An internal error has occurred.", mimetype="text/plain", status=400)	1✔
74	except BadRequest as ex:	1✔
75	current_app.logger.exception("%s, signature %s, path %s",str(ex), signature, path)	1✔
76	response = Response(f"An internal error has occurred.", mimetype="text/plain", status=400)	1✔
77	except Forbidden as ex:	1✔
78	current_app.logger.exception("%s, signature %s, path %s",str(ex), signature, path)	1✔
79	response = Response(f"An internal error has occurred.", mimetype="text/plain", status=403)	1✔
UNCOV 80	except Exception as ex:	×
81	current_app.logger.exception("%s, signature %s, path %s", str(ex), signature, path)	×
82	response = Response(f"Failed loading requested url.", mimetype="text/plain", status=400)	×
83	return response	1✔

bloomberg / pybossa / 19114618549

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous