legacy: support legacy secret links

site/zenodo_rdm/legacy/ext.py

@@ -7,6 +7,7 @@
 
 """Zenodo legacy API."""
 
+from flask_principal import identity_loaded
 from invenio_rdm_records.services.pids import PIDManager, PIDsService
 from invenio_rdm_records.services.review.service import ReviewService
 
@@ -24,6 +25,13 @@
     LegacyRecordService,
     LegacyRecordServiceConfig,
 )
+from .tokens import verify_legacy_secret_link
+
+
+@identity_loaded.connect
+def on_identity_loaded(_, identity):
+    """Add legacy secret link token to the freshly loaded Identity."""
+    verify_legacy_secret_link(identity)
 
 
 class ZenodoLegacy:

site/zenodo_rdm/legacy/tokens.py

@@ -0,0 +1,147 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2023 CERN.
+#
+# Zenodo is free software; you can redistribute it and/or modify
+# it under the terms of the MIT License; see LICENSE file for more details.
+
+"""Legacy token serializers.
+
+Ported from <https://github.com/zenodo/zenodo-accessrequests/blob/master/zenodo_accessrequests/tokens.py>.
+"""
+
+from collections import namedtuple
+from datetime import datetime
+from functools import partial
+
+from flask import current_app, flash, request, session
+from invenio_i18n import _
+from itsdangerous import BadData, SignatureExpired
+from itsdangerous.jws import JSONWebSignatureSerializer, TimedJSONWebSignatureSerializer
+
+_Need = namedtuple("Need", ["method", "value"])
+LegacySecretLinkNeed = partial(_Need, "legacy_secret_link")
+
+
+SUPPORTED_DIGEST_ALGORITHMS = ("HS256", "HS512")
+
+
+class TokenMixin:
+    """Mix-in class for token serializers."""
+
+    def validate_token(self, token, expected_data=None):
+        """Validate secret link token.
+
+        :param token: Token value.
+        :param expected_data: A dictionary of key/values that must be present
+            in the data part of the token.
+        """
+        try:
+            # Load token and remove random data.
+            data = self.load_token(token)
+
+            # Compare expected data with data in token.
+            if expected_data:
+                for k in expected_data:
+                    if expected_data[k] != data["data"].get(k):
+                        return None
+            return data
+        except BadData:
+            return None
+
+    def load_token(self, token, force=False):
+        """Load data in a token.
+
+        :param token: Token to load.
+        :param force: Load token data even if signature expired.
+                      Default: False.
+        """
+        try:
+            data = self.loads(token)
+        except SignatureExpired as e:
+            if not force:
+                raise
+            data = e.payload
+
+        del data["rnd"]
+        return data
+
+
+class SecretLinkSerializer(JSONWebSignatureSerializer, TokenMixin):
+    """Serializer for secret links."""
+
+    def __init__(self, **kwargs):
+        """Initialize underlying JSONWebSignatureSerializer."""
+        super(SecretLinkSerializer, self).__init__(
+            current_app.config["SECRET_KEY"], salt="accessrequests-link", **kwargs
+        )
+
+
+class TimedSecretLinkSerializer(TimedJSONWebSignatureSerializer, TokenMixin):
+    """Serializer for expiring secret links."""
+
+    def __init__(self, expires_at=None, **kwargs):
+        """Initialize underlying TimedJSONWebSignatureSerializer."""
+        assert isinstance(expires_at, datetime) or expires_at is None
+
+        dt = expires_at - datetime.now() if expires_at else None
+
+        super(TimedSecretLinkSerializer, self).__init__(
+            current_app.config["SECRET_KEY"],
+            expires_in=int(dt.total_seconds()) if dt else None,
+            salt="accessrequests-timedlink",
+            **kwargs
+        )
+
+
+class SecretLinkFactory:
+    """Functions for validating any secret link tokens."""
+
+    @classmethod
+    def validate_token(cls, token, expected_data=None):
+        """Validate a secret link token (non-expiring + expiring)."""
+        for algorithm in SUPPORTED_DIGEST_ALGORITHMS:
+            s = SecretLinkSerializer(algorithm_name=algorithm)
+            st = TimedSecretLinkSerializer(algorithm_name=algorithm)
+
+            try:
+                for serializer in (s, st):
+                    data = serializer.validate_token(token, expected_data=expected_data)
+                    if data:
+                        return data
+            except SignatureExpired:  # move to next algorithm
+                raise
+            except BadData:
+                continue  # move to next serializer/algorithm
+
+    @classmethod
+    def load_token(cls, token, force=False):
+        """Validate a secret link token (non-expiring + expiring)."""
+        for algorithm in SUPPORTED_DIGEST_ALGORITHMS:
+            s = SecretLinkSerializer(algorithm_name=algorithm)
+            st = TimedSecretLinkSerializer(algorithm_name=algorithm)
+            for serializer in (s, st):
+                try:
+                    data = serializer.load_token(token, force=force)
+                    if data:
+                        return data
+                except SignatureExpired:
+                    raise  # Signature was parsed and is expired
+                except BadData:
+                    continue  # move to next serializer/algorithm
+
+
+def verify_legacy_secret_link(identity):
+    """Verify the legacy secret linlk token."""
+    token_arg = "token"
+    session_arg = "_legacy_secret_link_token"
+    token = request.args.get(token_arg, session.get(session_arg, None))
+
+    if token:
+        try:
+            data = SecretLinkFactory.load_token(token)
+            if data:
+                identity.provides.add(LegacySecretLinkNeed(str(data["data"]["recid"])))
+                session[session_arg] = token
+        except SignatureExpired:
+            flash(_("Your shared link has expired."))

site/zenodo_rdm/permissions.py

@@ -22,14 +22,30 @@
     SubmissionReviewer,
 )
 from invenio_rdm_records.services.permissions import RDMRecordPermissionPolicy
-from invenio_records_permissions.generators import Disable, IfConfig, SystemProcess
+from invenio_records_permissions.generators import (
+    Disable,
+    Generator,
+    IfConfig,
+    SystemProcess,
+)
 from invenio_users_resources.services.permissions import UserManager
 
 from .generators import (
     IfFilesRestrictedForCommunity,
     IfRecordManagementAllowedForCommunity,
     MediaFilesManager,
 )
+from .legacy.tokens import LegacySecretLinkNeed
+
+
+class LegacySecretLinks(Generator):
+    """Legacy secret Links for records."""
+
+    def needs(self, record=None, **kwargs):
+        """Set of Needs granting permission."""
+        if record is None:
+            return []
+        return [LegacySecretLinkNeed(record["id"])]
 
 
 class ZenodoRDMRecordPermissionPolicy(RDMRecordPermissionPolicy):
@@ -49,6 +65,7 @@ class ZenodoRDMRecordPermissionPolicy(RDMRecordPermissionPolicy):
     can_preview = can_curate + [
         AccessGrant("preview"),
         SecretLinks("preview"),
+        LegacySecretLinks(),
         SubmissionReviewer(),
     ]
     can_view = can_preview + [