From cb8ec2a245f2ecc62296e99e5279099c621ee965 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Z=C3=BCbeyde=20Civelek?= Date: Mon, 18 Aug 2025 16:39:08 +0200 Subject: [PATCH 1/2] auth: allow external accounts to login and restrict upload --- cds/config.py | 2 +- cds/modules/deposit/views.py | 3 + cds/modules/home/views.py | 3 + cds/modules/invenio_deposit/utils.py | 2 + cds/modules/invenio_deposit/views/ui.py | 3 + cds/modules/ldap/decorators.py | 17 ++ cds/modules/oauthclient/cern_openid.py | 25 +- cds/modules/records/permissions.py | 11 +- scripts/setup | 3 + setup.cfg | 1 + tests/unit/conftest.py | 23 ++ tests/unit/test_external_user.py | 372 ++++++++++++++++++++++++ 12 files changed, 455 insertions(+), 10 deletions(-) create mode 100644 tests/unit/test_external_user.py diff --git a/cds/config.py b/cds/config.py index fa87fde27..5a31302a0 100644 --- a/cds/config.py +++ b/cds/config.py @@ -1173,7 +1173,7 @@ def _parse_env_bool(var_name, default=None): "https://auth.cern.ch/auth/realms/cern/protocol/openid-connect/userinfo", ) -OAUTHCLIENT_CERN_OPENID_ALLOWED_ROLES = ["cern-user"] +OAUTHCLIENT_CERN_OPENID_ALLOWED_ROLES = ["cern-user", "authenticated-user"] OAUTHCLIENT_CERN_OPENID_REFRESH_TIMEDELTA = timedelta(minutes=-5) """Default interval for refreshing CERN extra data (e.g. groups). diff --git a/cds/modules/deposit/views.py b/cds/modules/deposit/views.py index 1d880ed65..da8b9a7c9 100644 --- a/cds/modules/deposit/views.py +++ b/cds/modules/deposit/views.py @@ -25,6 +25,7 @@ """CDS interface.""" +from cds.modules.ldap.decorators import require_upload_permission from flask import ( Blueprint, abort, @@ -118,6 +119,7 @@ def to_links_js(pid, deposit=None, dep_type=None): @blueprint.route("/deposit/reportnumbers/new", methods=["GET", "POST"]) @login_required +@require_upload_permission() def reserve_report_number(): """Form to reserver a new report number.""" if not has_read_record_eos_path_permission(current_user, None): @@ -156,6 +158,7 @@ def reserve_report_number(): "/deposit/reportnumbers/assign/", methods=["GET", "POST"] ) @login_required +@require_upload_permission() def assign_report_number(depid): """Form to reserver a new report number.""" if not has_read_record_eos_path_permission(current_user, None): diff --git a/cds/modules/home/views.py b/cds/modules/home/views.py index 9a70122db..968d341f0 100644 --- a/cds/modules/home/views.py +++ b/cds/modules/home/views.py @@ -25,6 +25,8 @@ from invenio_cache.decorators import cached_unless_authenticated from invenio_i18n import lazy_gettext as _ +from ..records.permissions import has_upload_permission + blueprint = Blueprint( "cds_home", __name__, @@ -58,4 +60,5 @@ def init_menu(app): "invenio_deposit_ui.index", _("Upload"), order=2, + visible_when=lambda: has_upload_permission() ) diff --git a/cds/modules/invenio_deposit/utils.py b/cds/modules/invenio_deposit/utils.py index 71b22e05d..1db2d374d 100644 --- a/cds/modules/invenio_deposit/utils.py +++ b/cds/modules/invenio_deposit/utils.py @@ -28,6 +28,7 @@ from flask import request from invenio_oauth2server import require_api_auth, require_oauth_scopes +from cds.modules.ldap.decorators import require_upload_permission from .scopes import write_scope @@ -84,6 +85,7 @@ def check_oauth2_scope(can_method, *myscopes): def check(record, *args, **kwargs): @require_api_auth() + @require_upload_permission() @require_oauth_scopes(*myscopes) def can(self): return can_method(record) diff --git a/cds/modules/invenio_deposit/views/ui.py b/cds/modules/invenio_deposit/views/ui.py index 2d04b33d7..062202326 100644 --- a/cds/modules/invenio_deposit/views/ui.py +++ b/cds/modules/invenio_deposit/views/ui.py @@ -27,6 +27,7 @@ from copy import deepcopy +from cds.modules.ldap.decorators import require_upload_permission from flask import Blueprint, current_app, render_template, request from flask_login import login_required from invenio_pidstore.errors import PIDDeletedError @@ -73,12 +74,14 @@ def tombstone_errorhandler(error): @blueprint.route("/deposit") @login_required + @require_upload_permission() def index(): """List user deposits.""" return render_template(current_app.config["DEPOSIT_UI_INDEX_TEMPLATE"]) @blueprint.route("/deposit/new") @login_required + @require_upload_permission() def new(): """Create new deposit.""" deposit_type = request.values.get("type") diff --git a/cds/modules/ldap/decorators.py b/cds/modules/ldap/decorators.py index d2056e252..28e15d556 100644 --- a/cds/modules/ldap/decorators.py +++ b/cds/modules/ldap/decorators.py @@ -33,3 +33,20 @@ def decorated_api_view(*args, **kwargs): abort(401) return func(*args, **kwargs) return decorated_api_view + + +def require_upload_permission(): + """Restrict access using the has_upload_permission check.""" + def decorator(f): + from cds.modules.records.permissions import has_upload_permission + @wraps(f) + def decorated_function(*args, **kwargs): + if not current_user.is_authenticated: + abort(401) + + if not has_upload_permission(): + abort(403) + + return f(*args, **kwargs) + return decorated_function + return decorator diff --git a/cds/modules/oauthclient/cern_openid.py b/cds/modules/oauthclient/cern_openid.py index ae14b7d85..56af1ab73 100644 --- a/cds/modules/oauthclient/cern_openid.py +++ b/cds/modules/oauthclient/cern_openid.py @@ -85,8 +85,10 @@ def find_remote_by_client_id(client_id): def fetch_extra_data(resource): """Return a dict with extra data retrieved from CERN OAuth.""" - person_id = resource.get("cern_person_id") - return dict(person_id=person_id, groups=resource["groups"]) + data = {"groups": resource.get("groups", [])} + if resource.get("cern_person_id"): + data["person_id"] = resource["cern_person_id"] + return data def account_roles_and_extra_data(account, resource, refresh_timedelta=None): @@ -178,10 +180,19 @@ def _account_info(remote, resp): resp, ) - email = resource["email"] - external_id = str(resource["cern_uid"]) - nice = resource["preferred_username"] - name = resource["name"] + email = resource.get("email") + if not email: + raise OAuthCERNRejectedAccountError("No email in userinfo", remote, resp) + + external_id = resource.get("cern_uid") or resource.get("sub") + if not external_id: + raise OAuthCERNRejectedAccountError("No external_id in userinfo", remote, resp) + external_id = str(external_id) + raw_username = resource.get("preferred_username") or email + if "@" in raw_username: + raw_username = raw_username.replace("@", "_").replace(".", "_") + nice = raw_username + name = resource.get("name") or nice return dict( user=dict(email=email.lower(), profile=dict(username=nice, full_name=name)), @@ -231,7 +242,7 @@ def account_setup(remote, token, resp): resource = get_resource(remote, resp) with db.session.begin_nested(): - external_id = resource.get("cern_uid") + external_id = resource.get("cern_uid") or resource.get("sub") # Set CERN person ID in extra_data. token.remote_account.extra_data = {"external_id": external_id} diff --git a/cds/modules/records/permissions.py b/cds/modules/records/permissions.py index 295b7a0cb..44e1446c6 100644 --- a/cds/modules/records/permissions.py +++ b/cds/modules/records/permissions.py @@ -25,7 +25,7 @@ from flask import current_app from flask_security import current_user -from invenio_access import Permission +from invenio_access import Permission, action_factory from invenio_files_rest.models import Bucket, MultipartObject, ObjectVersion from invenio_records_files.api import FileObject from invenio_records_files.models import RecordsBuckets @@ -35,6 +35,8 @@ from .utils import get_user_provides, is_deposit, is_record, lowercase_value +upload_access_action = action_factory("videos-upload-access") + def files_permission_factory(obj, action=None): """Permission for files are always based on the type of bucket. @@ -228,7 +230,7 @@ def can(self): def create(cls, record, action, user=None): """Create a record permission.""" if action in cls.create_actions: - return cls(record, allow, user) + return cls(record, has_upload_permission, user) elif action in cls.read_actions: return cls(record, has_read_record_permission, user) elif action in cls.read_eos_path_actions: @@ -359,3 +361,8 @@ def has_admin_permission(user=None, record=None): """ # Allow administrators return Permission(action_admin_access).can() + + +def has_upload_permission(*args, **kwargs): + """Return permission to allow only cern users.""" + return Permission(upload_access_action).can() \ No newline at end of file diff --git a/scripts/setup b/scripts/setup index 434b5458b..075ae5ead 100755 --- a/scripts/setup +++ b/scripts/setup @@ -41,9 +41,12 @@ cds users create test@test.ch -a --password=123456 # Create an admin user cds users create admin@test.ch -a --password=123456 cds roles create admin +cds roles create cern-user +cds roles add test@test.ch cern-user cds roles add admin@test.ch admin cds access allow deposit-admin-access role admin cds access allow superuser-access role admin +cds access allow videos-upload-access role cern-user # Create a default files location cds files location --default videos /tmp/files diff --git a/setup.cfg b/setup.cfg index c8754551f..783559058 100644 --- a/setup.cfg +++ b/setup.cfg @@ -225,6 +225,7 @@ invenio_oauth2server.scopes = deposit_actions = cds.modules.invenio_deposit.scopes:actions_scope invenio_access.actions = deposit_admin_access = cds.modules.invenio_deposit.permissions:action_admin_access + upload_access_action = cds.modules.records.permissions:upload_access_action invenio_db.models = cds_migration_models = cds.modules.legacy.models diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index 7bf85c96b..a4a69fe43 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -79,6 +79,7 @@ from cds.modules.invenio_deposit.permissions import action_admin_access from cds.modules.records.resolver import record_resolver from cds.modules.redirector.views import api_blueprint as cds_api_blueprint +from cds.modules.records.permissions import upload_access_action @pytest.yield_fixture(scope="module", autouse=True) @@ -203,6 +204,15 @@ def users(app, db): superadmin_role = Role(name="superadmin") db.session.add(ActionRoles(action=superuser_access.value, role=superadmin_role)) datastore.add_role_to_user(superadmin, superadmin_role) + # Give upload permission to all users + cern_user_role = Role(name="cern-user") + db.session.add( + ActionRoles(action=upload_access_action.value, role=cern_user_role) + ) + datastore.add_role_to_user(admin, cern_user_role) + datastore.add_role_to_user(user1, cern_user_role) + datastore.add_role_to_user(user2, cern_user_role) + datastore.add_role_to_user(superadmin, cern_user_role) db.session.commit() id_1 = user1.id id_2 = user2.id @@ -210,6 +220,19 @@ def users(app, db): return [id_1, id_2, id_4] +@pytest.fixture() +def external_user(app, db): + """Create external user.""" + with db.session.begin_nested(): + datastore = app.extensions["security"].datastore + user = datastore.create_user( + email="external@gmail.com", password="tester", active=True + ) + db.session.commit() + id = user.id + return id + + @pytest.fixture() def u_email(db, users): """Valid user email.""" diff --git a/tests/unit/test_external_user.py b/tests/unit/test_external_user.py new file mode 100644 index 000000000..8f15cbc7e --- /dev/null +++ b/tests/unit/test_external_user.py @@ -0,0 +1,372 @@ +# -*- coding: utf-8 -*- +# +# This file is part of CDS. +# Copyright (C) 2025 CERN. +# +# CDS is free software; you can redistribute it +# and/or modify it under the terms of the GNU General Public License as +# published by the Free Software Foundation; either version 2 of the +# License, or (at your option) any later version. +# +# CDS 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 +# General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with CDS; if not, write to the +# Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, +# MA 02111-1307, USA. +# +# In applying this license, CERN does not +# waive the privileges and immunities granted to it by virtue of its status +# as an Intergovernmental Organization or submit itself to any jurisdiction. + + +"""Tests for external user permissions.""" + +import json +from io import BytesIO + +from flask import url_for +from flask_principal import AnonymousIdentity, UserNeed, identity_loaded +from flask_security import current_user, login_user, logout_user +from helpers import prepare_videos_for_publish +from invenio_access import Permission +from invenio_access.models import ActionRoles +from invenio_accounts.models import Role, User + +from cds.modules.deposit.api import deposit_video_resolver +from cds.modules.records.permissions import ( + has_upload_permission, + record_permission_factory, + upload_access_action, +) + + +def test_has_upload_permission_external_user(app, external_user): + """Test that external user without cern-user role cannot upload.""" + with app.test_request_context(): + user = User.query.get(external_user) + login_user(user) + + # Test has_upload_permission function directly + permission = Permission(upload_access_action) + assert not permission.can() + + # Test has_upload_permission helper function + assert not has_upload_permission() + + +def test_has_upload_permission_cern_user(app, users): + """Test that authenticated user with cern-user role can upload.""" + with app.test_request_context(): + user = User.query.get(users[0]) + login_user(user) + + # Test has_upload_permission function directly + permission = Permission(upload_access_action) + assert permission.can() + + # Test has_upload_permission helper function + assert has_upload_permission() + + +def test_record_create_permission_external_user(app, external_user, deposit_metadata): + """Test record create permission for external user without role.""" + with app.test_request_context(): + user = User.query.get(external_user) + login_user(user) + + # Test creating a record permission + factory = record_permission_factory(record=deposit_metadata, action="create") + assert not factory.can() + + +def test_project_rest_api_external_user_can_create( + api_app, external_user, deposit_metadata, json_partial_project_headers +): + """Test project creation via REST API for external user without role.""" + with api_app.test_client() as client: + user = User.query.get(external_user) + login_user(user) + + # Try to create a project via REST API + resp = client.post( + url_for("invenio_deposit_rest.project_list"), + data=json.dumps(deposit_metadata), + headers=json_partial_project_headers, + ) + # Should be forbidden (403) because user doesn't have upload permission + assert resp.status_code == 403 + + +def test_video_rest_api_external_user( + api_app, external_user, video_deposit_metadata, json_partial_project_headers +): + """Test video creation via REST API for external user without role.""" + with api_app.test_client() as client: + user = User.query.get(external_user) + login_user(user) + + # Try to create a video via REST API + resp = client.post( + url_for("invenio_deposit_rest.video_list"), + data=json.dumps(video_deposit_metadata), + headers=json_partial_project_headers, + ) + # Should be forbidden (403) because user doesn't have upload permission + assert resp.status_code == 403 + + +def test_anonymous_user_has_upload_permission(app): + """Test that anonymous users cannot upload.""" + with app.test_request_context(): + # No user logged in + logout_user() + + # Test has_upload_permission function directly + permission = Permission(upload_access_action) + assert not permission.can() + + # Test has_upload_permission helper function + assert not has_upload_permission() + + +def test_external_user_role_assignment(app, db, external_user): + """Test that we can dynamically add cern-user role to external user.""" + with app.test_request_context(): + user = User.query.get(external_user) + login_user(user) + + # Initially should not have upload permission + assert not has_upload_permission() + + # Add cern-user role + datastore = app.extensions["security"].datastore + cern_user_role = Role.query.filter_by(name="cern-user").first() + if not cern_user_role: + cern_user_role = Role(name="cern-user") + db.session.add( + ActionRoles(action=upload_access_action.value, role=cern_user_role) + ) + datastore.add_role_to_user(user, cern_user_role) + db.session.commit() + + # Need to logout and login again for role to take effect + logout_user() + login_user(user) + + # Now should have upload permission + assert has_upload_permission() + + +def test_published_video_access_control_external_user( + api_app, location, users, external_user, api_project +): + """Test external user access to published video records.""" + + @identity_loaded.connect + def mock_identity_provides(sender, identity): + """Ensure external users have their email in identity for testing.""" + if ( + not isinstance(identity, AnonymousIdentity) + and current_user.is_authenticated + ): + # Add UserNeed with email for all authenticated users (including external users) + if ( + current_user.email + and UserNeed(current_user.email) not in identity.provides + ): + identity.provides.add(UserNeed(current_user.email)) + + (_, video_1, video_2) = api_project + cern_user = User.query.filter_by(id=users[0]).first() + user2 = User.query.filter_by(id=users[1]).first() + ext_user = User.query.filter_by(id=external_user).first() + + # Prepare videos for publishing + prepare_videos_for_publish([video_1, video_2]) + vid1 = video_1["_deposit"]["id"] + vid2 = video_2["_deposit"]["id"] + + with api_app.test_client() as client: + login_user(cern_user) + + # Create restricted video (user2 access only) + video_1_metadata = dict(video_1) + for key in ["_files"]: + video_1_metadata.pop(key, None) + video_1_metadata["_access"] = {"read": [user2.email]} + + resp = client.put( + url_for("invenio_deposit_rest.video_item", pid_value=vid1), + data=json.dumps(video_1_metadata), + headers=[ + ("Content-Type", "application/vnd.video.partial+json"), + ("Accept", "application/json"), + ], + ) + assert resp.status_code == 200 + + # Publish restricted video + url = url_for( + "invenio_deposit_rest.video_actions", pid_value=vid1, action="publish" + ) + assert client.post(url).status_code == 202 + rec_pid1, _ = deposit_video_resolver(vid1).fetch_published() + + # Create restricted video (external user access only) + video_2_metadata = dict(video_2) + for key in ["_files"]: + video_2_metadata.pop(key, None) + video_2_metadata["_access"] = {"read": [ext_user.email]} + + resp = client.put( + url_for("invenio_deposit_rest.video_item", pid_value=vid2), + data=json.dumps(video_2_metadata), + headers=[ + ("Content-Type", "application/vnd.video.partial+json"), + ("Accept", "application/json"), + ], + ) + assert resp.status_code == 200 + + # Publish restricted video (external user access only) + url = url_for( + "invenio_deposit_rest.video_actions", pid_value=vid2, action="publish" + ) + assert client.post(url).status_code == 202 + rec_pid2, _ = deposit_video_resolver(vid2).fetch_published() + + # Test external user access + logout_user() + login_user(ext_user) + + # External user should be blocked from video1 + resp1 = client.get( + url_for("invenio_records_rest.recid_item", pid_value=rec_pid1.pid_value) + ) + assert resp1.status_code in [403, 404] + + # External user should access video2 + resp2 = client.get( + url_for("invenio_records_rest.recid_item", pid_value=rec_pid2.pid_value) + ) + assert resp2.status_code == 200 + video_data = json.loads(resp2.data.decode("utf-8")) + assert "metadata" in video_data + + +def test_external_user_deposit_operations( + api_app, + location, + external_user, + users, + deposit_metadata, + project_deposit_metadata, + video_deposit_metadata, + json_partial_project_headers, + json_partial_video_headers, +): + """Tests for external user deposit operations and permissions.""" + with api_app.test_request_context(): + # Setup: Create project and video as CERN user + cern_user = User.query.get(users[0]) + login_user(cern_user) + + with api_app.test_client() as client: + # Create project + resp = client.post( + url_for("invenio_deposit_rest.project_list"), + data=json.dumps(project_deposit_metadata), + headers=json_partial_project_headers, + ) + assert resp.status_code == 201 + project_data = json.loads(resp.data.decode("utf-8")) + project_id = project_data["metadata"]["_deposit"]["id"] + + # Create video + video_deposit_metadata["_project_id"] = project_id + resp = client.post( + url_for("invenio_deposit_rest.video_list"), + data=json.dumps(video_deposit_metadata), + headers=json_partial_video_headers, + ) + assert resp.status_code == 201 + video_data = json.loads(resp.data.decode("utf-8")) + video_id = video_data["metadata"]["_deposit"]["id"] + + # Switch to external user for testing + logout_user() + ext_user = User.query.get(external_user) + login_user(ext_user) + + # Test 1: Project creation - should be forbidden + resp = client.post( + url_for("invenio_deposit_rest.project_list"), + data=json.dumps(deposit_metadata), + headers=json_partial_project_headers, + ) + assert resp.status_code == 403 + + # Test 2: Project item operations - should be forbidden + # GET project + resp = client.get( + url_for("invenio_deposit_rest.project_item", pid_value=project_id) + ) + assert resp.status_code in [403, 404] + + # PUT project + resp = client.put( + url_for("invenio_deposit_rest.project_item", pid_value=project_id), + data=json.dumps(deposit_metadata), + headers=json_partial_project_headers, + ) + assert resp.status_code == 403 + + # DELETE project + resp = client.delete( + url_for("invenio_deposit_rest.project_item", pid_value=project_id) + ) + assert resp.status_code == 403 + + # Test 3: Project actions - should be forbidden + actions = ["publish", "edit", "discard"] + for action in actions: + resp = client.post( + url_for( + "invenio_deposit_rest.project_actions", + pid_value=project_id, + action=action, + ) + ) + assert resp.status_code in [403, 404] + + # Test 4: File operations - should be forbidden + # GET files list + resp = client.get( + url_for("invenio_deposit_rest.project_files", pid_value=project_id) + ) + assert resp.status_code in [403, 404] + + # POST file upload + resp = client.post( + url_for("invenio_deposit_rest.project_files", pid_value=project_id), + data={"file": (BytesIO(b"test content"), "test.txt")}, + ) + assert resp.status_code in [403, 404] + + # Test 5: Flows API - should be forbidden + flow_payload = { + "bucket_id": "test-bucket-id", + "deposit_id": video_id, + "key": "test-file.mp4", + "version_id": "test-version-id", + } + resp = client.post( + "/api/flows/", + data=json.dumps(flow_payload), + headers=json_partial_project_headers, + ) + assert resp.status_code in [401, 403, 404] From e59cef47ee48682a043a841d8e2b1c776042bba0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Z=C3=BCbeyde=20Civelek?= Date: Tue, 9 Sep 2025 14:59:29 +0200 Subject: [PATCH 2/2] permissions: require upload permission for edit --- cds/modules/records/permissions.py | 3 + tests/unit/test_external_user.py | 105 ++++++++++++++++++++++++++--- 2 files changed, 97 insertions(+), 11 deletions(-) diff --git a/cds/modules/records/permissions.py b/cds/modules/records/permissions.py index 44e1446c6..9a9b37137 100644 --- a/cds/modules/records/permissions.py +++ b/cds/modules/records/permissions.py @@ -336,6 +336,9 @@ def has_update_permission(user, record): """Check if user has update access to the record.""" user_id = int(user.get_id()) if user.is_authenticated else None + if not has_upload_permission(): + return False + # Allow owners deposit_creator = record.get("_deposit", {}).get("created_by", -1) if user_id == deposit_creator: diff --git a/tests/unit/test_external_user.py b/tests/unit/test_external_user.py index 8f15cbc7e..0ffbe20a8 100644 --- a/tests/unit/test_external_user.py +++ b/tests/unit/test_external_user.py @@ -274,7 +274,7 @@ def test_external_user_deposit_operations( # Setup: Create project and video as CERN user cern_user = User.query.get(users[0]) login_user(cern_user) - + with api_app.test_client() as client: # Create project resp = client.post( @@ -285,7 +285,7 @@ def test_external_user_deposit_operations( assert resp.status_code == 201 project_data = json.loads(resp.data.decode("utf-8")) project_id = project_data["metadata"]["_deposit"]["id"] - + # Create video video_deposit_metadata["_project_id"] = project_id resp = client.post( @@ -296,12 +296,12 @@ def test_external_user_deposit_operations( assert resp.status_code == 201 video_data = json.loads(resp.data.decode("utf-8")) video_id = video_data["metadata"]["_deposit"]["id"] - + # Switch to external user for testing logout_user() ext_user = User.query.get(external_user) login_user(ext_user) - + # Test 1: Project creation - should be forbidden resp = client.post( url_for("invenio_deposit_rest.project_list"), @@ -309,14 +309,14 @@ def test_external_user_deposit_operations( headers=json_partial_project_headers, ) assert resp.status_code == 403 - + # Test 2: Project item operations - should be forbidden # GET project resp = client.get( url_for("invenio_deposit_rest.project_item", pid_value=project_id) ) assert resp.status_code in [403, 404] - + # PUT project resp = client.put( url_for("invenio_deposit_rest.project_item", pid_value=project_id), @@ -324,13 +324,13 @@ def test_external_user_deposit_operations( headers=json_partial_project_headers, ) assert resp.status_code == 403 - + # DELETE project resp = client.delete( url_for("invenio_deposit_rest.project_item", pid_value=project_id) ) assert resp.status_code == 403 - + # Test 3: Project actions - should be forbidden actions = ["publish", "edit", "discard"] for action in actions: @@ -342,21 +342,21 @@ def test_external_user_deposit_operations( ) ) assert resp.status_code in [403, 404] - + # Test 4: File operations - should be forbidden # GET files list resp = client.get( url_for("invenio_deposit_rest.project_files", pid_value=project_id) ) assert resp.status_code in [403, 404] - + # POST file upload resp = client.post( url_for("invenio_deposit_rest.project_files", pid_value=project_id), data={"file": (BytesIO(b"test content"), "test.txt")}, ) assert resp.status_code in [403, 404] - + # Test 5: Flows API - should be forbidden flow_payload = { "bucket_id": "test-bucket-id", @@ -370,3 +370,86 @@ def test_external_user_deposit_operations( headers=json_partial_project_headers, ) assert resp.status_code in [401, 403, 404] + + +def test_external_user_update_access_without_upload_permission( + api_app, location, users, external_user, api_project +): + """Test that external user in _access.update still can't edit without upload permission.""" + + @identity_loaded.connect + def mock_identity_provides(sender, identity): + """Ensure external users have their email in identity for testing.""" + if ( + not isinstance(identity, AnonymousIdentity) + and current_user.is_authenticated + ): + if ( + current_user.email + and UserNeed(current_user.email) not in identity.provides + ): + identity.provides.add(UserNeed(current_user.email)) + + (_, video_1, _) = api_project + cern_user = User.query.get(users[0]) + ext_user = User.query.get(external_user) + + # Prepare videos for publishing + prepare_videos_for_publish([video_1]) + vid1 = video_1["_deposit"]["id"] + + with api_app.test_client() as client: + login_user(cern_user) + + # Create video with external user in update access + video_1_metadata = dict(video_1) + for key in ["_files"]: + video_1_metadata.pop(key, None) + video_1_metadata["_access"]["update"] = [ext_user.email] + + resp = client.put( + url_for("invenio_deposit_rest.video_item", pid_value=vid1), + data=json.dumps(video_1_metadata), + headers=[ + ("Content-Type", "application/vnd.video.partial+json"), + ("Accept", "application/json"), + ], + ) + # Publish video + url = url_for( + "invenio_deposit_rest.video_actions", pid_value=vid1, action="publish" + ) + assert client.post(url).status_code == 202 + rec_pid1, _ = deposit_video_resolver(vid1).fetch_published() + + # Test external user (has update access but can't edit) + logout_user() + login_user(ext_user) + + # External user should be able to read the record + resp = client.get( + url_for("invenio_records_rest.recid_item", pid_value=rec_pid1.pid_value) + ) + assert resp.status_code == 200 + video_data = json.loads(resp.data.decode("utf-8")) + + project_id = video_data["metadata"]["_project_id"] + deposit_id = video_data["metadata"]["_deposit"]["id"] + + # External user should not be able to get the deposit + resp = client.get( + url_for("invenio_deposit_rest.project_item", pid_value=project_id) + ) + assert resp.status_code in [403, 404] + + # External user should not be able to get the video deposit + res = client.get( + url_for("invenio_deposit_rest.video_item", pid_value=deposit_id) + ) + assert res.status_code in [403, 404] + + # External user should not be able to edit the video + url = url_for( + "invenio_deposit_rest.video_actions", pid_value=deposit_id, action="edit" + ) + assert client.post(url).status_code in [403, 404]