diff --git a/backend/ibutsu_server/controllers/artifact_controller.py b/backend/ibutsu_server/controllers/artifact_controller.py index 68f8a52d..2e47d0bd 100644 --- a/backend/ibutsu_server/controllers/artifact_controller.py +++ b/backend/ibutsu_server/controllers/artifact_controller.py @@ -27,6 +27,20 @@ def __init__(self, message, status=HTTPStatus.BAD_REQUEST): self.status = status +def _user_has_artifact_access(artifact, user): + """Check whether the user has access to the artifact's project. + + Access is determined by the project associated with the artifact's result or run. + If the artifact is not linked to either a result or a run, access is denied + to avoid silently granting access to orphaned artifacts. + """ + if artifact.result: + return project_has_user(artifact.result.project, user) + if artifact.run: + return project_has_user(artifact.run.project, user) + return False + + def _get_form_param(form, *keys): """Get a form parameter supporting both camelCase and snake_case naming. @@ -68,9 +82,9 @@ def view_artifact(id_, token_info=None, user=None): :rtype: file """ artifact, response = _build_artifact_response(id_) - if (artifact.result and not project_has_user(artifact.result.project, user)) or ( - artifact.run and not project_has_user(artifact.run.project, user) - ): + if not isinstance(artifact, Artifact): + return artifact, response + if not _user_has_artifact_access(artifact, user): return HTTPStatus.FORBIDDEN.phrase, HTTPStatus.FORBIDDEN return response @@ -85,7 +99,9 @@ def download_artifact(id_, token_info=None, user=None): :rtype: file """ artifact, response = _build_artifact_response(id_) - if not project_has_user(artifact.result.project, user): + if not isinstance(artifact, Artifact): + return artifact, response + if not _user_has_artifact_access(artifact, user): return HTTPStatus.FORBIDDEN.phrase, HTTPStatus.FORBIDDEN response.headers["Content-Disposition"] = f"attachment; filename={artifact.filename}" return response @@ -103,7 +119,7 @@ def get_artifact(id_, token_info=None, user=None): artifact = db.session.get(Artifact, id_) if not artifact: return HTTPStatus.NOT_FOUND.phrase, HTTPStatus.NOT_FOUND - if not project_has_user(artifact.result.project, user): + if not _user_has_artifact_access(artifact, user): return HTTPStatus.FORBIDDEN.phrase, HTTPStatus.FORBIDDEN return artifact.to_dict() @@ -120,8 +136,6 @@ def get_artifact_list( """ query = db.select(Artifact) user = db.session.get(User, user) - if "result_id" in request.args: - result_id = request.args["result_id"] if result_id: query = query.where(Artifact.result_id == result_id) if run_id: @@ -338,7 +352,7 @@ def delete_artifact(id_, token_info=None, user=None): artifact = db.session.get(Artifact, id_) if not artifact: return HTTPStatus.NOT_FOUND.phrase, HTTPStatus.NOT_FOUND - if not project_has_user(artifact.result.project, user): + if not _user_has_artifact_access(artifact, user): return HTTPStatus.FORBIDDEN.phrase, HTTPStatus.FORBIDDEN session.delete(artifact) session.commit() diff --git a/backend/ibutsu_server/controllers/import_controller.py b/backend/ibutsu_server/controllers/import_controller.py index deea7135..606dd4b4 100644 --- a/backend/ibutsu_server/controllers/import_controller.py +++ b/backend/ibutsu_server/controllers/import_controller.py @@ -102,7 +102,7 @@ def add_import( project_obj = get_project(project) if not project_obj: return f"Project {project} doesn't exist", HTTPStatus.BAD_REQUEST - if not project_has_user(project, user): + if not project_has_user(project_obj, user): return HTTPStatus.FORBIDDEN.phrase, HTTPStatus.FORBIDDEN data["project_id"] = project_obj.id if request.form.get("metadata"): diff --git a/backend/ibutsu_server/controllers/login_controller.py b/backend/ibutsu_server/controllers/login_controller.py index d68edf80..ea645484 100644 --- a/backend/ibutsu_server/controllers/login_controller.py +++ b/backend/ibutsu_server/controllers/login_controller.py @@ -291,7 +291,7 @@ def recover(body=None): if not user: return HTTPStatus.BAD_REQUEST.phrase, HTTPStatus.BAD_REQUEST # Create a random activation code. Base64 just for funsies - user.activation_code = urlsafe_b64encode(str(uuid4()).encode("utf8")).strip(b"=") + user.activation_code = urlsafe_b64encode(str(uuid4()).encode("utf8")).strip(b"=").decode() session.add(user) session.commit() return {}, HTTPStatus.CREATED diff --git a/backend/ibutsu_server/util/projects.py b/backend/ibutsu_server/util/projects.py index 471daf42..d32c56fd 100644 --- a/backend/ibutsu_server/util/projects.py +++ b/backend/ibutsu_server/util/projects.py @@ -28,6 +28,8 @@ def project_has_user(project, user): return True if isinstance(project, str): project = get_project(project) + if project is None: + return False return user in project.users or project.owner.id == user.id diff --git a/backend/tests/test_util.py b/backend/tests/test_util.py index 18a39297..1069343a 100644 --- a/backend/tests/test_util.py +++ b/backend/tests/test_util.py @@ -560,6 +560,18 @@ def test_project_has_user(make_project, make_user, user_role, expected_result): assert result is expected_result +def test_project_has_user_none_project(make_user): + """Test project_has_user returns False when project is None (data integrity issue).""" + user = make_user(email="user@test.com") + assert project_has_user(None, user) is False + + +def test_project_has_user_invalid_string_project_id(make_user): + """Test project_has_user returns False when a string project ID resolves to None.""" + user = make_user(email="user@test.com") + assert project_has_user("non-existent-project-id", user) is False + + def test_project_has_user_with_string_ids(make_project, make_user): """Test project_has_user with string IDs instead of objects.""" owner = make_user(email="owner@test.com")