Merge pull request #11 from BorjaEst/10-flaat-v110-upgrade

10 flaat v110 upgrade

.env-example

@@ -16,11 +16,11 @@ DB_NAME=eosc_perf
 BCRYPT_LOG_ROUNDS=12
 
 ## Authorization configuration.
+# TRUSTED_OP_LIST=https://aai.egi.eu/oidc
 OIDC_CLIENT_ID=eosc-performance
 OIDC_CLIENT_SECRET=some-secret-token
 # OIDC_CLIENT_SECRET_FILE=path/to/oidc/secret
 ADMIN_ENTITLEMENTS=urn:something:somewhere
-DISABLE_ADMIN_PROTECTION=false
 
 ## Email and notification configuration.
 [email protected]

backend/VERSION

@@ -1 +1 @@
-1.0.0
+1.1.0

backend/app.py

@@ -10,7 +10,7 @@
 
 from . import routes
 from .extensions import api         # Api interface module
-from .extensions import auth        # flaat ext. manage db migrations
+from .extensions import flaat       # Flask authentication with tokens
 from .extensions import db          # SQLAlchemy instance
 from .extensions import migrate     # Alembic ext. manage db migrations
 from .extensions import mail        # Mail ext. to send notifications
@@ -84,7 +84,7 @@ def register_extensions(app):
     api.init_app(app)
     db.init_app(app)
     migrate.init_app(app, db)
-    auth.init_app(app)
+    flaat.init_app(app)
     mail.init_app(app)
 
 

backend/authorization.py

@@ -1,214 +1,26 @@
-"""Authorization module."""
-# Simplify code when Authlib V1.0 using:
-# https://docs.authlib.org/en/latest/specs/rfc7662.html
-#   #use-introspection-in-resource-server
+from flaat.config import AccessLevel
+from flaat.requirements import IsTrue
+from flask import current_app
 
-from functools import wraps
+from backend import models
 
-from aarc_g002_entitlement import (Aarc_g002_entitlement,
-                                   Aarc_g002_entitlement_Error,
-                                   Aarc_g002_entitlement_ParseError)
-from flaat import Flaat, tokentools
-from flask import current_app, request
-from flask_smorest import abort
 
-from . import models
+def is_registered(user_infos):
+    """Assert user is registered in the database."""
+    user = models.User.read((user_infos.subject, user_infos.issuer))
+    return user is not None
 
 
-class Authorization(Flaat):
-    """Monkeypatch flaat to solve lazy configuration
-    https://github.com/indigo-dc/flaat/issues/32
+def is_admin(user_infos):
+    """Assert registration and entitlements."""
+    entitlements = set(user_infos.user_info['eduperson_entitlement'])
+    return all([
+        entitlements & set(current_app.config['ADMIN_ENTITLEMENTS']),
+        is_registered(user_infos),
+    ])
 
-    For more information see:
-    https://flask.palletsprojects.com/en/2.0.x/extensiondev/#the-extension-code
-    """
 
-    def __init__(self, app=None):
-        self.app = app
-        self.admin_entitlements = None
-        if app is not None:
-            self.init_app(app)
-
-    def init_app(self, app):
-        super().__init__()
-
-        self.set_web_framework('flask')
-        self.set_trusted_OP_list([
-            'https://aai.egi.eu/oidc',
-            'https://aai-demo.egi.eu/auth/realms/egi',
-            'https://aai-dev.egi.eu/auth/realms/egi',
-        ])
-
-        # Flaat timeout:
-        timeout = app.config.get('FLAAT_TIMEOUT', 3)
-        self.set_timeout(timeout)
-
-        # verbosity:
-        #     0: No output
-        #     1: Errors
-        #     2: More info, including token info
-        #     3: Max
-        verbosity = app.config.get('FLAAT_VERBOSITY', 0)
-        self.set_verbosity(verbosity)
-
-        # Required for using token introspection endpoint:
-        client_id = app.config['OIDC_CLIENT_ID']
-        self.set_client_id(client_id)
-
-        client_secret = app.config['OIDC_CLIENT_SECRET']
-        self.set_client_secret(client_secret)
-
-        admin_entitlements = app.config['ADMIN_ENTITLEMENTS']
-        self.admin_entitlements = admin_entitlements
-
-    def current_tokeninfo(self):
-        """Returns the token information from the current request.
-
-        :return: Token information.
-        :rtype: dict or None
-        """
-        token = tokentools.get_access_token_from_request(request)
-        info = tokentools.get_accesstoken_info(token)
-        return info['body'] if 'body' in info else None
-
-    def current_userinfo(self):
-        """Returns the token user info from the introspection endpoint.
-
-        :return: User introspection endpoint information.
-        :rtype: dict or None
-        """
-        token = tokentools.get_access_token_from_request(request)
-        user_info = self.get_all_info_by_at(token)
-        return user_info
-
-    def valid_token(self):
-        """Function to evaluate the validity of the user login"""
-        try:
-            all_info = self._get_all_info_from_request(request)
-            current_app.logger.debug(f"request info: {all_info}")
-            return all_info is not None
-
-        except Exception as e:
-            current_app.logger.error('Error validating user', exc_info=e)
-            return False
-
-    def token_required(self, on_failure=None):
-        """Decorator to enforce a valid login.
-        Optional on_failure is called if no valid user detected.
-        Useful for redirecting to some login page"""
-        def wrapper(view_func):
-            @wraps(view_func)
-            def decorated(*args, **kwargs):
-                if self.valid_token():
-                    current_app.logger.debug("Token accepted")
-                    return self._wrap_async_call(view_func, *args, **kwargs)
-                elif on_failure:
-                    failure = on_failure(self.get_last_error())
-                    return self._return_formatter_wf(failure, 401)
-                else:
-                    alert = f"No valid authentication: {self.get_last_error()}"
-                    abort(401, message=alert)
-            return decorated
-        return wrapper
-
-    def valid_user(self):
-        """Function to evaluate the validity of the user login"""
-        user = models.User.current_user()
-        if not user:
-            current_app.logger.error("User not registered")
-            return False
-        else:
-            current_app.logger.debug(f"User info: {user}")
-            return True
-
-    def login_required(self, on_failure=None):
-        """Decorator to enforce a valid login.
-        Optional on_failure is called if no valid user detected.
-        Useful for redirecting to some login page"""
-        def wrapper(view_func):
-            @wraps(view_func)
-            def decorated(*args, **kwargs):
-                if self.valid_user():
-                    current_app.logger.debug("User accepted")
-                    return self._wrap_async_call(view_func, *args, **kwargs)
-                elif on_failure:
-                    failure = on_failure(self.get_last_error())
-                    return self._return_formatter_wf(failure, 403)
-                else:
-                    abort(403, messages={'user': "Not registered"})
-            return self.token_required()(decorated)
-        return wrapper
-
-    def valid_admin(self, match='all'):
-        """Function to evaluate the validity of the user as admin"""
-        if current_app.config['DISABLE_ADMIN_PROTECTION']:
-            current_app.logger.warning("ADMIN validation is disabled")
-            return True
-
-        try:
-            claim = 'eduperson_entitlement'
-            all_info = self._get_all_info_from_request(request)
-            current_app.logger.debug(f"request info: {all_info}")
-
-            group = self.admin_entitlements
-            req_glist = group if isinstance(group, list) else [group]
-
-            # copy entries from incoming claim
-            (g_entries, user_msg) = self._get_entitlements_from_claim(
-                all_info, claim)
-            if not g_entries:
-                return False
-
-            required = self._determine_number_of_required_matches(
-                match, req_glist)
-            if not required:
-                raise Exception("Error interpreting 'match' parameter")
-
-            def e_expander(es):
-                """Helper function to catch exceptions in list comprehension"""
-                try:
-                    return Aarc_g002_entitlement(es, strict=False)
-                except ValueError:
-                    return None
-                except Aarc_g002_entitlement_ParseError:
-                    return None
-                except Aarc_g002_entitlement_Error:
-                    return None
-
-            avail_entitlements = [e_expander(
-                es) for es in g_entries if e_expander(es) is not None]
-            req_entitlements = [e_expander(
-                es) for es in req_glist if e_expander(es) is not None]
-
-            # now we do the actual checking
-            matches = 0
-            for req in req_entitlements:
-                for avail in avail_entitlements:
-                    if req.is_contained_in(avail):
-                        matches += 1
-
-            current_app.logger.debug(f"found {matches} of {required} matches")
-            return matches >= required
-
-        except Exception as e:
-            current_app.logger.error('Error validating admin', exc_info=e)
-            return False
-
-    def admin_required(self, on_failure=None):
-        """Decorator to enforce a valid admin.
-        Optional on_failure is called if no valid admin detected.
-        Useful for redirecting to some login page"""
-        def wrapper(view_func):
-            @wraps(view_func)
-            def decorated(*args, **kwargs):
-                if self.valid_admin():
-                    current_app.logger.debug("Admin accepted")
-                    return self._wrap_async_call(view_func, *args, **kwargs)
-                elif on_failure:
-                    alert = 'You are not authorized'
-                    return self._return_formatter_wf(on_failure(alert), 403)
-                else:
-                    alert = 'You are not authorized'
-                    return self._return_formatter_wf(alert, 403)
-            return self.login_required()(decorated)
-        return wrapper
+access_levels = [
+    AccessLevel("user", IsTrue(is_registered)),
+    AccessLevel("admin", IsTrue(is_admin)),
+]

backend/extensions.py

@@ -8,15 +8,16 @@
 lately initialized in the application factory using the settings and
 configurations from the environment.
 """
+from flaat.flask import Flaat
 from flask_mailman import Mail
 from flask_migrate import Migrate
 from flask_smorest import Api
 from flask_sqlalchemy import SQLAlchemy
 
-from .authorization import Authorization
+from backend import authorization
 
 #: Flask extension that provides support for handling oidc Access Tokens
-auth = Authorization()
+flaat = Flaat(authorization.access_levels)
 
 #: Flask framework library for creating REST APIs (i.e. OpenAPI)
 api = Api()

backend/models/models/reports/claim.py

@@ -105,15 +105,18 @@ def claims(cls):
             backref=backref("resource", uselist=False),
         )
 
-    def claim(self, message):
+    def claim(self, claimer, message):
         """Creates a pending claim related to the resource and soft
         deletes the resource.
 
+        :param claimer: Message to include in the claim
+        :type claimer: models.User
         :param message: Message to include in the claim
         :type message: str
         """
         self.delete()
         return self.__class__._claim_report_class(
+            uploader=claimer,
             message=message,
             resource=self
         )

backend/models/models/user.py

@@ -5,7 +5,6 @@
 from sqlalchemy.ext.declarative import declared_attr
 from sqlalchemy.orm import backref, relationship
 
-from ...extensions import auth
 from ..core import TokenModel
 
 
@@ -39,11 +38,6 @@ def __repr__(self) -> str:
         """Human-readable representation string"""
         return "<{} {}>".format(self.__class__.__name__, self.email)
 
-    @classmethod
-    def current_user(cls):
-        tokeninfo = auth.current_tokeninfo()
-        return cls.read((tokeninfo['sub'], tokeninfo['iss']))
-
 
 class HasUploader(object):
     """Mixin that adds an User as upload details to any model."""
@@ -60,25 +54,9 @@ class HasUploader(object):
     #: (ISO8601) Upload datetime of the model instance
     upload_datetime = Column(DateTime, nullable=False, default=dt.now)
 
-    def __init__(self, **properties):
-        super().__init__(**properties)
-        if self.uploader is None:
-            self.uploader = User.current_user()
-
     @declared_attr
     def uploader(cls):
         """(User class) User that uploaded the model instance"""
         return relationship("User", backref=backref(
             f'_{cls.__name__.lower()}s', cascade="all, delete-orphan"
         ))
-
-    def update(self, *args, force=False, **kwargs):
-        """Decorates super() that only owner can edit unless force=True.
-        """
-        if force or self.ownership():
-            super().update(*args, **kwargs)
-        else:
-            raise PermissionError
-
-    def ownership(self):
-        return self.uploader == User.current_user()