feat(harvester): use dedicated user with inspire-harvester role

site/cds_rdm/administration/harvester_reports.py

@@ -158,6 +158,5 @@ def init_search_config(self, **kwargs):
             default_size=20,
             hidden_params=[
                 ["action", "record.publish"],
-                ["user_id", "system"],
             ],
         )

site/cds_rdm/config.py

@@ -64,3 +64,6 @@
 
 This only applies to records where the specified community is their parent's default community.
 """
+
+CDS_HARVESTER_USER_EMAIL = None
+"""Email of the INSPIRE harvester service user."""

site/cds_rdm/generators.py

@@ -12,6 +12,7 @@
 from flask_principal import RoleNeed, UserNeed
 from invenio_access import action_factory
 from invenio_access.permissions import Permission
+from invenio_accounts.proxies import current_datastore
 from invenio_rdm_records.services.generators import AccessGrant
 from invenio_records_permissions.generators import AuthenticatedUser, Generator
 from invenio_search.engine import dsl
@@ -20,6 +21,7 @@
 
 archiver_read_all_role = RoleNeed("archiver-read-all")
 archiver_notification_role = RoleNeed("archiver-notification")
+inspire_harvester_role = RoleNeed("inspire-harvester")
 
 clc_sync_action = action_factory("clc-sync")
 clc_sync_permission = Permission(clc_sync_action)
@@ -68,7 +70,10 @@ class AuthenticatedRegularUser(AuthenticatedUser):
     def excludes(self, **kwargs):
         """Exclude service/robot accounts."""
         excludes = super().excludes(**kwargs)
-        return excludes + [archiver_read_all_role, archiver_notification_role]
+        return excludes + [
+            archiver_read_all_role,
+            archiver_notification_role,
+        ]
 
 
 class ArchiverRole(Generator):
@@ -110,6 +115,14 @@ def archiver_role(self):
         return archiver_notification_role
 
 
+class InspireHarvester(Generator):
+    """Allows by inspire-harvester role."""
+
+    def needs(self, **kwargs):
+        """Enabling Needs."""
+        return [inspire_harvester_role]
+
+
 class HarvesterCurator(Generator):
     """Allows harvester curators via the harvester admin action."""
 
@@ -118,12 +131,19 @@ def needs(self, **kwargs):
         return [harvester_admin_access_action]
 
     def query_filter(self, identity=None, **kwargs):
-        """Restrict harvester curators to system-user ``record.publish`` audit logs."""
+        """Filter to harvester and legacy system publish audit logs."""
         if identity and Permission(harvester_admin_access_action).allows(identity):
+            user_ids = ["system"]
+            email = current_app.config.get("CDS_HARVESTER_USER_EMAIL")
+            if email:
+                user = current_datastore.get_user_by_email(email)
+                if user is not None:
+                    user_ids.append(str(user.id))
+
             return dsl.Q(
                 "bool",
                 must=[
-                    dsl.Q("term", **{"user.id": "system"}),
+                    dsl.Q("terms", **{"user.id": user_ids}),
                     dsl.Q("term", action="record.publish"),
                 ],
             )

site/cds_rdm/inspire_harvester/load/draft.py

@@ -8,7 +8,6 @@
 """Draft lifecycle management module."""
 
 from flask import current_app
-from invenio_access.permissions import system_identity
 from invenio_db import db
 from invenio_rdm_records.proxies import current_rdm_records_service
 from invenio_rdm_records.services.errors import ValidationErrorWithMessageAsList
@@ -24,23 +23,27 @@
 class DraftLifecycleManager:
     """Manages draft creation, editing, versioning, and publishing."""
 
+    def __init__(self, identity):
+        """Constructor."""
+        self.identity = identity
+
     def create(self, entry):
         """Create a new draft from entry data."""
-        return current_rdm_records_service.create(system_identity, data=entry)
+        return current_rdm_records_service.create(self.identity, data=entry)
 
     def edit(self, record_pid):
         """Open an edit draft for an existing published record."""
-        return current_rdm_records_service.edit(system_identity, record_pid)
+        return current_rdm_records_service.edit(self.identity, record_pid)
 
     def update(self, draft, metadata):
         """Update draft."""
         return current_rdm_records_service.update_draft(
-            system_identity, draft.id, metadata
+            self.identity, draft.id, metadata
         )
 
     def new_version(self, record_pid):
         """Create a new-version draft from an existing published record."""
-        return current_rdm_records_service.new_version(system_identity, record_pid)
+        return current_rdm_records_service.new_version(self.identity, record_pid)
 
     def add_cern_research_community(self, draft):
         """Add the CERN Scientific Community to the draft."""
@@ -58,22 +61,22 @@ def delete_files(self, draft_id, filenames, logger):
         for filename in filenames:
             logger.debug(f"Delete file: {filename}")
             current_rdm_records_service.draft_files.delete_file(
-                system_identity, draft_id, filename
+                self.identity, draft_id, filename
             )
 
     def publish(self, draft_id, logger):
         """Publish a draft. Deletes the draft on any failure, then raises WriterError."""
         try:
             logger.debug(f"Publishing draft {draft_id}")
-            current_rdm_records_service.publish(system_identity, draft_id)
+            current_rdm_records_service.publish(self.identity, draft_id)
             logger.info(f"Draft {draft_id} published successfully.")
         except (ValidationError, ValidationErrorWithMessageAsList) as e:
-            current_rdm_records_service.delete_draft(system_identity, draft_id)
+            current_rdm_records_service.delete_draft(self.identity, draft_id)
             raise WriterError(
                 f"Record validation failed: {format_validation_error(e)}"
             ) from e
         except Exception as e:
-            current_rdm_records_service.delete_draft(system_identity, draft_id)
+            current_rdm_records_service.delete_draft(self.identity, draft_id)
             raise_unexpected_operation_error(
                 subject="draft",
                 action="published",

site/cds_rdm/inspire_harvester/load/files.py

@@ -14,7 +14,6 @@
 from typing import List
 
 import requests
-from invenio_access.permissions import system_identity
 from invenio_rdm_records.proxies import current_rdm_records_service
 from invenio_records_resources.services.errors import FileKeyNotFoundError
 from invenio_vocabularies.datastreams.errors import WriterError
@@ -46,10 +45,12 @@ def __init__(
         self,
         retry_config: RetryConfig = None,
         draft_lifecycle: DraftLifecycleManager = None,
+        identity=None,
     ):
         """Constructor."""
         self.retry_config = retry_config or RetryConfig()
-        self.draft_lifecycle = draft_lifecycle or DraftLifecycleManager()
+        self.draft_lifecycle = draft_lifecycle
+        self.identity = identity
 
     def _populate_missing_checksums(self, files, logger):
         """Calculate checksums for incoming files that do not provide one."""
@@ -153,7 +154,7 @@ def sync(self, draft, record, incoming_record, logger, import_files=True):
             record_dict = record.to_dict()
             existing_files = record_dict["files"]["entries"]
         if should_import_files:
-            current_rdm_records_service.import_files(system_identity, draft.id)
+            current_rdm_records_service.import_files(self.identity, draft.id)
             logger.debug(
                 f"Imported files to {draft.id} from previous version: {record.id}"
             )
@@ -197,19 +198,19 @@ def _upload_file(self, draft, file_data, file_content, logger):
             }
             logger.debug(f"Filename: '{file_data['key']}' initializing.")
             service.draft_files.init_files(
-                system_identity, draft.id, [file_data_to_init]
+                self.identity, draft.id, [file_data_to_init]
             )
             logger.debug(f"Filename: '{file_data['key']}' initialized successfully.")
 
             service.draft_files.set_file_content(
-                system_identity, draft.id, file_data["key"], file_content
+                self.identity, draft.id, file_data["key"], file_content
             )
             logger.debug(
                 f"Filename: '{file_data['key']}' content set successfully. Commit file..."
             )
 
             result = service.draft_files.commit_file(
-                system_identity, draft.id, file_data["key"]
+                self.identity, draft.id, file_data["key"]
             )
             new_checksum = result.data["checksum"]
             logger.debug(
@@ -224,5 +225,5 @@ def _upload_file(self, draft, file_data, file_content, logger):
                 f"Files checksums don't match."
                 f" Delete file: '{file_data['key']}' from draft."
             )
-            service.draft_files.delete_file(system_identity, draft.id, file_data["key"])
+            service.draft_files.delete_file(self.identity, draft.id, file_data["key"])
             raise WriterError("File checksum mismatch.")

site/cds_rdm/inspire_harvester/load/matcher.py

@@ -10,7 +10,6 @@
 from dataclasses import dataclass, field
 from typing import List, Optional
 
-from invenio_access.permissions import system_identity
 from invenio_rdm_records.proxies import current_rdm_records_service
 from invenio_search.engine import dsl
 
@@ -121,6 +120,10 @@ def query(self):
 class RecordMatcher:
     """Finds existing CDS records that match an incoming INSPIRE entry."""
 
+    def __init__(self, identity):
+        """Constructor."""
+        self.identity = identity
+
     def match(self, stream_entry, inspire_id, logger) -> MatchResult:
         """Search for existing records using a priority-ordered filter chain."""
         entry = stream_entry.entry
@@ -134,7 +137,7 @@ def match(self, stream_entry, inspire_id, logger) -> MatchResult:
                 combined_filter = dsl.Q("bool", filter=candidate.query)
                 logger.debug(f"Searching for existing records: {candidate.query}")
                 result = current_rdm_records_service.search(
-                    system_identity, extra_filter=combined_filter
+                    self.identity, extra_filter=combined_filter
                 )
                 if result.total >= 1:
                     logger.debug(f"Found {result.total} matching records.")

site/cds_rdm/inspire_harvester/transform/transform_entry.py

@@ -9,7 +9,6 @@
 from copy import deepcopy
 
 from flask import current_app
-from invenio_access.permissions import system_user_id
 
 from cds_rdm.inspire_harvester.logger import Logger
 from cds_rdm.inspire_harvester.transform.config import mapper_policy
@@ -22,10 +21,11 @@
 class RDMEntry:
     """Building of CDS-RDM entry record."""
 
-    def __init__(self, inspire_record):
+    def __init__(self, inspire_record, harvester_user):
         """Initializes the RDM entry."""
         self.inspire_record = inspire_record
         self.inspire_metadata = inspire_record["metadata"]
+        self.harvester_user = harvester_user
         self.transformer = Inspire2RDM(self.inspire_record)
         self.cds_id = self.transformer.cds_id
         self.splitter = InspireVersionSplitter(
@@ -55,7 +55,7 @@ def _parent(self):
         parent = {
             "access": {
                 "owned_by": {
-                    "user": system_user_id,
+                    "user": self.harvester_user.id,
                 }
             }
         }

site/cds_rdm/inspire_harvester/transformer.py

@@ -7,6 +7,8 @@
 
 """Transformer module."""
 from flask import current_app
+from invenio_access.permissions import system_identity
+from invenio_users_resources.proxies import current_users_service
 from invenio_vocabularies.datastreams.transformers import BaseTransformer
 
 from .transform.transform_entry import RDMEntry
@@ -18,14 +20,18 @@ class InspireJsonTransformer(BaseTransformer):
     def __init__(self, root_element=None, *args, **kwargs):
         """Initializes the transformer."""
         self.root_element = root_element
+        email = current_app.config["CDS_HARVESTER_USER_EMAIL"]
+        self.harvester_user = current_users_service.read_by_email(
+            system_identity, email
+        )
         super().__init__(*args, **kwargs)
 
     def apply(self, stream_entry, **kwargs):
         """Applies the transformation to the INSPIRE record entry."""
         current_app.logger.info("Start transformation of INSPIRE record to CDS record.")
         # assign original source record to the stream entry
         stream_entry.source_entry = stream_entry.entry
-        entry_builder = RDMEntry(stream_entry.entry)
+        entry_builder = RDMEntry(stream_entry.entry, self.harvester_user)
         rdm_entry, versions, cds_id, errors = entry_builder.build()
 
         if errors:

site/cds_rdm/inspire_harvester/writer.py

@@ -11,7 +11,9 @@
 
 from flask import current_app
 from invenio_access.permissions import system_identity
+from invenio_access.utils import get_identity
 from invenio_rdm_records.proxies import current_rdm_records_service
+from invenio_users_resources.proxies import current_users_service
 from invenio_vocabularies.datastreams.errors import WriterError
 from invenio_vocabularies.datastreams.writers import BaseWriter
 from marshmallow import ValidationError
@@ -42,9 +44,14 @@ class InspireWriter(BaseWriter):
 
     def __init__(self):
         """Constructor."""
-        self.matcher = RecordMatcher()
-        self.drafts = DraftLifecycleManager()
-        self.file_sync = FileSynchronizer(draft_lifecycle=self.drafts)
+        email = current_app.config["CDS_HARVESTER_USER_EMAIL"]
+        user = current_users_service.read_by_email(system_identity, email)
+        self.identity = get_identity(user._user.model.model_obj)
+        self.matcher = RecordMatcher(self.identity)
+        self.drafts = DraftLifecycleManager(self.identity)
+        self.file_sync = FileSynchronizer(
+            draft_lifecycle=self.drafts, identity=self.identity
+        )
 
     def write(self, stream_entry, *args, **kwargs):
         """Create or update the record in CDS."""
@@ -114,7 +121,7 @@ def _update_record(
         """Dispatch to in-place edit or new-version based on file/DOI state."""
         entry = {k: v for k, v in stream_entry.entry.items() if k != "_inspire_ctx"}
         ctx = stream_entry.entry["_inspire_ctx"]
-        record = current_rdm_records_service.read(system_identity, record_pid)
+        record = current_rdm_records_service.read(self.identity, record_pid)
         record_dict = record.to_dict()
 
         should_update_files = self.file_sync.check_files_should_update(
@@ -172,7 +179,7 @@ def _update_record(
     def _resource_type_versioning(self, record, update_metadata, ctx, logger):
 
         search_result = current_rdm_records_service.scan_versions(
-            identity=system_identity,
+            identity=self.identity,
             id_=record.id,
         )
         existing_record_versions = {
@@ -187,7 +194,7 @@ def _resource_type_versioning(self, record, update_metadata, ctx, logger):
             logger.info(f"Processing {incoming_resource_type} version")
             if incoming_resource_type in existing_record_versions:
                 version_record = current_rdm_records_service.read(
-                    system_identity, existing_record_versions[incoming_resource_type]
+                    self.identity, existing_record_versions[incoming_resource_type]
                 )
                 should_update_files = self.file_sync.check_files_should_update(
                     version_record, version, logger
@@ -205,10 +212,14 @@ def _resource_type_versioning(self, record, update_metadata, ctx, logger):
             else:
                 self._publish_new_version(record, version, logger)
 
-        latest_record_version = current_rdm_records_service.record_cls.get_latest_published_by_parent(
-            record._record.parent
+        latest_record_version = (
+            current_rdm_records_service.record_cls.get_latest_published_by_parent(
+                record._record.parent
+            )
+        )
+        record = current_rdm_records_service.read(
+            self.identity, latest_record_version["id"]
         )
-        record = current_rdm_records_service.read(system_identity, latest_record_version["id"])
         # publish the latest version at the end
         self._publish_new_version(record, update_metadata, logger)
         logger.info(
@@ -282,7 +293,7 @@ def _create_record(
 
             self.drafts.add_cern_research_community(draft)
         except Exception:
-            current_rdm_records_service.delete_draft(system_identity, draft.id)
+            current_rdm_records_service.delete_draft(self.identity, draft.id)
             logger.error(f"Draft {draft.id} is deleted due to errors.")
             raise