Implement e-kirjasto authentication flow for admins

alembic/versions/20240222_9966f6f95674_create_admincredentials_table.py

@@ -0,0 +1,38 @@
+"""Create admincredentials table
+
+Revision ID: 9966f6f95674
+Revises: 993729d4bf97
+Create Date: 2024-02-22 02:36:07.130941+00:00
+
+"""
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "9966f6f95674"
+down_revision = "993729d4bf97"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "admincredentials",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("external_id", sa.Unicode(), nullable=False),
+        sa.Column("admin_id", sa.Integer(), nullable=False),
+        sa.ForeignKeyConstraint(["admin_id"], ["admins.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        op.f("ix_admincredentials_admin_id"),
+        "admincredentials",
+        ["admin_id"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(op.f("ix_admincredentials_admin_id"), table_name="admincredentials")
+    op.drop_table("admincredentials")

api/admin/config.py

@@ -3,7 +3,10 @@
 from urllib.parse import urljoin
 
 from requests import RequestException
+from sqlalchemy.engine.url import make_url
+from sqlalchemy.exc import ArgumentError
 
+from core.config import CannotLoadConfiguration
 from core.util.http import HTTP, RequestNetworkException
 from core.util.log import LoggerMixin
 
@@ -50,9 +53,31 @@ class Configuration(LoggerMixin):
     ENV_ADMIN_UI_PACKAGE_NAME = "TPP_CIRCULATION_ADMIN_PACKAGE_NAME"
     ENV_ADMIN_UI_PACKAGE_VERSION = "TPP_CIRCULATION_ADMIN_PACKAGE_VERSION"
 
+    # Finland
+    # Environment variable that defines admin ekirjasto sign in URL
+    ENV_ADMIN_EKIRJASTO_AUTHENTICATION_URL = "ADMIN_EKIRJASTO_AUTHENTICATION_URL"
+
     # Cache the package version after first lookup.
     _version: str | None = None
 
+    @classmethod
+    def ekirjasto_authentication_url(cls) -> str:
+        url = os.environ.get(cls.ENV_ADMIN_EKIRJASTO_AUTHENTICATION_URL)
+        if not url:
+            raise CannotLoadConfiguration(
+                "Admin Ekirjasto authentication url was not defined in environment variable %s."
+                % cls.ENV_ADMIN_EKIRJASTO_AUTHENTICATION_URL
+            )
+
+        try:
+            make_url(url)
+        except ArgumentError as e:
+            raise ArgumentError(
+                "Bad format for admin ekirjasto authentication URL (%s)." % url
+            )
+
+        return url
+
     @classmethod
     def operational_mode(cls) -> OperationalMode:
         return (

api/admin/controller/base.py

@@ -6,6 +6,9 @@
 
 import flask
 
+from api.admin.ekirjasto_admin_authentication_provider import (
+    EkirjastoAdminAuthenticationProvider,
+)
 from api.admin.exceptions import AdminNotAuthorized
 from api.admin.password_admin_authentication_provider import (
     PasswordAdminAuthenticationProvider,
@@ -35,8 +38,10 @@ def __init__(self, manager):
     @property
     def admin_auth_providers(self):
         if Admin.with_password(self._db).count() != 0:
-            return [PasswordAdminAuthenticationProvider()]
-
+            return [
+                EkirjastoAdminAuthenticationProvider(),
+                PasswordAdminAuthenticationProvider(),
+            ]
         return []
 
     def admin_auth_provider(self, type):

api/admin/controller/individual_admin_settings.py

@@ -19,6 +19,7 @@
 )
 from api.problem_details import LIBRARY_NOT_FOUND
 from core.model import Admin, AdminRole, Library, get_one, get_one_or_create
+from core.model.admin import AdminCredential
 from core.util.problem_detail import ProblemDetail
 
 
@@ -68,14 +69,25 @@ def process_get(self):
         if not highest_role:
             raise AdminNotAuthorized()
 
+        # Finland, don't allow externally authenticated users to manage admins.
+        if logged_in_admin.is_authenticated_externally():
+            raise AdminNotAuthorized()
+
         def append_role(roles, role):
             role_dict = dict(role=role.role)
             if role.library:
                 role_dict["library"] = role.library.short_name
             roles.append(role_dict)
 
         admins = []
-        for admin in self._db.query(Admin).order_by(Admin.email):
+        for admin in (
+            self._db.query(Admin)
+            .outerjoin(
+                AdminCredential
+            )  # Finland, don't return externally authenticated admins
+            .filter(AdminCredential.admin_id.is_(None))
+            .order_by(Admin.email)
+        ):
             roles = []
             show_admin = True
             for role in admin.roles:
@@ -271,6 +283,13 @@ def check_permissions(self, admin, settingUp):
         if not settingUp:
             user = flask.request.admin
 
+            # Finland, don't allow editing externally authenticated users
+            if (
+                user.is_authenticated_externally()
+                or admin.is_authenticated_externally()
+            ):
+                raise AdminNotAuthorized()
+
             # System admin has all permissions.
             if user.is_system_admin():
                 return

api/admin/controller/reset_password.py

@@ -70,7 +70,8 @@ def forgot_password(self) -> ProblemDetail | WerkzeugResponse:
 
         admin = self._extract_admin_from_request(flask.request)
 
-        if not admin:
+        # Finland, extra check to disable password reset for ekirjasto authenticated users.
+        if not admin or admin.is_authenticated_externally():
             return self._response_with_message_and_redirect_button(
                 INVALID_ADMIN_CREDENTIALS.detail,
                 url_for("admin_forgot_password"),

api/admin/controller/sign_in.py

@@ -10,12 +10,17 @@
 
 from api.admin.config import Configuration as AdminClientConfig
 from api.admin.controller.base import AdminController
+from api.admin.ekirjasto_admin_authentication_provider import (
+    EkirjastoAdminAuthenticationProvider,
+    EkirjastoUserInfo,
+)
 from api.admin.password_admin_authentication_provider import (
     PasswordAdminAuthenticationProvider,
 )
 from api.admin.problem_details import (
     ADMIN_AUTH_MECHANISM_NOT_CONFIGURED,
     ADMIN_AUTH_NOT_CONFIGURED,
+    ADMIN_NOT_AUTHORIZED,
     INVALID_ADMIN_CREDENTIALS,
 )
 from api.admin.template_styles import (
@@ -26,6 +31,9 @@
     section_style,
     small_link_style,
 )
+from api.problem_details import EKIRJASTO_REMOTE_AUTHENTICATION_FAILED
+from core.model import get_one, get_one_or_create
+from core.model.admin import Admin, AdminCredential, AdminRole
 from core.util.problem_detail import ProblemDetail
 
 
@@ -67,6 +75,99 @@ class SignInController(AdminController):
         logo=logo_style,
     )
 
+    # Finland
+    def ekirjasto_auth_finish(self):
+        auth: EkirjastoAdminAuthenticationProvider = self.admin_auth_provider(
+            EkirjastoAdminAuthenticationProvider.NAME
+        )
+        if not auth:
+            return ADMIN_AUTH_MECHANISM_NOT_CONFIGURED
+
+        result = flask.request.form.get("result")
+        if result != "success":
+            logging.error("Ekirjasto authentication failed, result = %s", result)
+            return self.error_response(EKIRJASTO_REMOTE_AUTHENTICATION_FAILED)
+
+        ekirjasto_token = flask.request.form.get("token")
+
+        user_info = auth.ekirjasto_authenticate(ekirjasto_token)
+        if isinstance(user_info, ProblemDetail):
+            return user_info
+
+        circulation_role = self._to_circulation_role(user_info.role)
+        if not circulation_role:
+            return self.error_response(ADMIN_NOT_AUTHORIZED)
+
+        try:
+            credentials = get_one(self._db, AdminCredential, external_id=user_info.sub)
+            if credentials:
+                admin = credentials.admin
+            else:
+                admin = self._create_admin_with_external_credentials(user_info)
+
+            self._update_roles_if_changed(admin, circulation_role)
+        except Exception as e:
+            logging.exception("Internal error during signup")
+            self._db.rollback()
+            return EKIRJASTO_REMOTE_AUTHENTICATION_FAILED
+
+        self._setup_admin_flask_session(admin, auth, ekirjasto_token)
+
+        redirect_uri = flask.request.args.get("redirect_uri", "/admin/web")
+        return SanitizedRedirections.redirect(redirect_uri)
+
+    def _create_admin_with_external_credentials(self, user_info: EkirjastoUserInfo):
+        admin, _ = get_one_or_create(
+            self._db,
+            Admin,
+            email=user_info.sub,
+        )
+        get_one_or_create(
+            self._db,
+            AdminCredential,
+            external_id=user_info.sub,
+            admin_id=admin.id,
+        )
+        return admin
+
+    @staticmethod
+    def _update_roles_if_changed(admin: Admin, new_role: str):
+        existing_roles = [(role.role, role.library) for role in admin.roles]
+        if [(new_role, None)] != existing_roles:
+            for role in admin.roles:
+                admin.remove_role(role.role, role.library)
+            admin.add_role(new_role)
+
+    @staticmethod
+    def _setup_admin_flask_session(
+        admin: Admin,
+        auth: EkirjastoAdminAuthenticationProvider,
+        ekirjasto_token: str,
+    ):
+        # Set up the admin's flask session.
+        flask.session["admin_email"] = admin.email
+        flask.session["auth_type"] = auth.NAME
+
+        # This one is extra compared to password auth provider
+        flask.session["ekirjasto_token"] = ekirjasto_token
+
+        # A permanent session expires after a fixed time, rather than
+        # when the user closes the browser.
+        flask.session.permanent = True
+
+    @staticmethod
+    def _to_circulation_role(ekirjasto_role: str) -> str | None:
+        if ekirjasto_role == "orgadmin":
+            return AdminRole.SYSTEM_ADMIN
+        elif ekirjasto_role == "admin":
+            return AdminRole.SITEWIDE_LIBRARY_MANAGER
+        elif ekirjasto_role == "librarian":
+            return AdminRole.SITEWIDE_LIBRARIAN
+        else:
+            # other possible values are "sysadmin", "registrant" and "customer",
+            # these are not allowed as circulation admins
+            return None
+
     def sign_in(self):
         """Redirects admin if they're signed in, or shows the sign in page."""
         if not self.admin_auth_providers:
@@ -98,7 +199,9 @@ def sign_in(self):
             headers["Content-Type"] = "text/html"
             return Response(html, 200, headers)
         elif admin:
-            return SanitizedRedirections.redirect(flask.request.args.get("redirect"))
+            return SanitizedRedirections.redirect(
+                flask.request.args.get("redirect", "/admin/web")
+            )
 
     def password_sign_in(self):
         if not self.admin_auth_providers:
@@ -117,6 +220,10 @@ def password_sign_in(self):
 
     def change_password(self):
         admin = flask.request.admin
+
+        if admin.is_authenticated_externally():
+            return ADMIN_NOT_AUTHORIZED
+
         new_password = flask.request.form.get("password")
         if new_password:
             admin.password = new_password
@@ -127,13 +234,25 @@ def sign_out(self):
         flask.session.pop("admin_email", None)
         flask.session.pop("auth_type", None)
 
+        # Finland, revoke ekirjasto session
+        self._try_revoke_ekirjasto_session()
+
         redirect_url = url_for(
             "admin_sign_in",
             redirect=url_for("admin_view", _external=True),
             _external=True,
         )
         return SanitizedRedirections.redirect(redirect_url)
 
+    # Finland
+    def _try_revoke_ekirjasto_session(self):
+        ekirjasto_token = flask.session.pop("ekirjasto_token", None)
+        auth: EkirjastoAdminAuthenticationProvider = self.admin_auth_provider(
+            EkirjastoAdminAuthenticationProvider.NAME
+        )
+        if ekirjasto_token and auth:
+            auth.try_revoke_ekirjasto_session(ekirjasto_token)
+
     def error_response(self, problem_detail):
         """Returns a problem detail as an HTML response"""
         html = self.ERROR_RESPONSE_TEMPLATE % dict(