Task/2711 auth for admin

app/__init__.py

@@ -13,7 +13,9 @@
 from marshmallow.exceptions import ValidationError
 
 from app.auth.User import User, UserAnonymous
+from app.auth.Admin import Admin, AdminAnonymous
 from app.auth.service.VoucherUserService import VoucherUserService
+from app.auth.service.VoucherAdminService import VoucherAdminService
 from app.config.settings.ProcessSetting import process_config
 from app.config.settings.SystemSettings import SystemSettingsObj
 from app.config.init_config import init_get_system_and_default_domain_settings, init_get_user_domain_settings
@@ -137,19 +139,28 @@ def check_content_type() -> ResponseReturnValue | None:  # pylint: disable=too-m
     @base_blueprint.before_request
     def get_user() -> ResponseReturnValue | None:
         """
-        Add the user instance, even if there is no user
+        Add the user/admin instance to Flask g, even if there is no user/admin
         """
 
         auth_header = request.authorization
-        user: User = UserAnonymous()
+        user: User | Admin = UserAnonymous()
+        admin: Admin = AdminAnonymous()
+
         if auth_header:
             if auth_header.type == 'bearer':
-                user = VoucherUserService(process_config).generate_user_from_voucher(auth_header.token)
+                if kind == cs.API_BASIC:
+                    user = VoucherUserService(process_config).generate_user_from_voucher(auth_header.token)
+                elif kind == cs.API_ADMIN:
+                    admin = VoucherAdminService(process_config).generate_admin_from_voucher(auth_header.token)
             elif auth_header.type == 'basic' and current_app.config[cs.ALLOW_AUTH_BASIC]:
                 pass
             else:
                 return create_api_base_response(error=err.ERROR_WRONG_AUTHORIZATION_TYPE)
-        g.user = user
+
+        if kind == cs.API_BASIC:
+            g.user = user
+        elif kind == cs.API_ADMIN:
+            g.admin = admin
         return None
 
     if kind == cs.API_BASIC:
@@ -174,6 +185,27 @@ def check_non_anonymous_endpoint() -> ResponseReturnValue | None:
                 return create_api_base_response(error=err.ERROR_AUTHENTICATED_ROUTE)
             return None
 
+    if kind == cs.API_ADMIN:
+        @base_blueprint.before_request
+        def check_admin_authenticated() -> ResponseReturnValue | None:
+            """
+            Check that admin is authenticated for protected endpoints
+            """
+            # Skip authentication check for OPTIONS (CORS preflight)
+            if request.method == "OPTIONS":
+                return None
+
+            # Endpoints that don't require admin authentication
+            anon_admin_endpoints = {
+                "admin#AdminAuth.v1_AdminAuth.AdminAuth.ApiAdminAuthLogin",
+            }
+
+            if (isinstance(g.admin, AdminAnonymous)
+                    and request.endpoint not in anon_admin_endpoints
+                    and not _is_public_endpoint()):
+                return create_api_base_response(error=err.ERROR_AUTHENTICATED_ROUTE)
+            return None
+
     if sogo_state == cs.SOGO_NOT_INIT:
         if kind == cs.API_BASIC:
             @base_blueprint.before_request
@@ -188,7 +220,7 @@ def add_process() -> None:
                 """
                 _Add the process settings in g
                 """
-                if 'process' not in g:
+                if 'process_settings' not in g:
                     g.process_settings = process_config
 
     elif sogo_state == cs.SOGO_OK:
@@ -197,26 +229,35 @@ def get_config_and_user() -> ResponseReturnValue | None:
             """
             Get and set the config in the global flask
             """
-            if 'process' not in g:
+            if 'process_settings' not in g:
                 g.process_settings = process_config
             system_settings, default_domain_settings = init_get_system_and_default_domain_settings()
             if 'system_settings' not in g:
                 g.system_settings = system_settings
             if 'default_domain' not in g:
                 g.default_domain_settings = default_domain_settings
-            if 'user' in g:
-                user: User = g.user
-                if isinstance(user, UserAnonymous):
-                    g.user_domain_settings = default_domain_settings
+
+            # Handle basic API (user-based)
+            if kind == cs.API_BASIC:
+                if 'user' in g:
+                    user: User = g.user
+                    if isinstance(user, UserAnonymous):
+                        g.user_domain_settings = default_domain_settings
+                    else:
+                        g.user_domain_settings = init_get_user_domain_settings(user)
+                        inter = InterfaceAuthUser(process_config, system_settings, g.user_domain_settings)
+                        creds_ok = inter.check_user_and_fill_info(user)
+                        if not creds_ok:
+                            return create_api_base_response(error=err.ERROR_USER_CREDS_NOT_VALID)
                 else:
-                    g.user_domain_settings = init_get_user_domain_settings(user)
-                    inter = InterfaceAuthUser(process_config, system_settings, g.user_domain_settings)
-                    creds_ok = inter.check_user_and_fill_info(user)
-                    if not creds_ok:
-                        return create_api_base_response(error=err.ERROR_USER_CREDS_NOT_VALID)
-            else:
-                logger.error("No user in Flask g")
-                raise AggravatedException("No user in Flask g")
+                    logger.error("No user in Flask g")
+                    raise AggravatedException("No user in Flask g")
+            # Handle admin API (admin-based)
+            elif kind == cs.API_ADMIN:
+                if 'admin' not in g:
+                    logger.error("No admin in Flask g")
+                    raise AggravatedException("No admin in Flask g")
+
             return None
 
 

app/api/v1/admin/ApiAdminAuth.py

@@ -0,0 +1,70 @@
+"""
+Admin authentication API endpoints
+"""
+
+from __future__ import annotations
+from typing import TYPE_CHECKING
+
+from flask import g, request, Response
+from flask.views import MethodView
+from flask.typing import ResponseReturnValue
+from flask_smorest import Blueprint
+
+from app.interface.admin.InterfaceApiAdminAuth import InterfaceAdminAuth
+from app.utils.logger.logger import logger_api
+
+from .schema import adminAuth as sch
+
+if TYPE_CHECKING:
+    from app.config.settings.ProcessSetting import ProcessSetting
+
+
+blp = Blueprint("AdminAuth", __name__, url_prefix="/auth")
+
+
+@blp.before_request
+def init_admin_auth_interface() -> None:
+    """
+    Initialize the admin auth interface for this request
+    """
+    logger_api.debug("Calling before_request for AdminAuth")
+    process: ProcessSetting = g.process_settings
+    interface_api = InterfaceAdminAuth(process)
+    g.inter = interface_api
+
+
+@blp.route("/login")
+class ApiAdminAuthLogin(MethodView):
+    """
+    Admin login endpoint
+    """
+
+    @blp.arguments(sch.AdminAuthBasicPostSchema, example=sch.AdminAuthBasicPostSchema.example(), error_status_code=400)
+    @blp.response(200)
+    def post(self, new_data: dict) -> ResponseReturnValue:
+        """
+        Authenticate admin with username and password.
+        Returns a JWT token for subsequent requests.
+        """
+        username = new_data["username"]
+        password = new_data["password"]
+        interface_api: InterfaceAdminAuth = g.inter
+        return interface_api.admin_login(username, password)
+
+
+@blp.route("/logout")
+class ApiAdminAuthLogout(MethodView):
+    """
+    Admin logout endpoint
+    """
+
+    @blp.response(200)
+    def post(self) -> ResponseReturnValue:
+        """
+        Logout the authenticated admin by revoking the session associated
+        with the JWT token present in the Authorization header.
+        """
+        auth_header = request.authorization
+        voucher_data: str = auth_header.token if auth_header else ""
+        interface_api: InterfaceAdminAuth = g.inter
+        return interface_api.admin_logout(voucher_data)

app/api/v1/admin/__init__.py

@@ -2,5 +2,6 @@
 
 from .ApiAdminConfig import blp as admin_config_api_blueprint
 from .ApiAdminUser import blp as admin_user_api_blueprint
+from .ApiAdminAuth import blp as admin_auth_api_blueprint
 
-admin_apis : list[Blueprint] = [admin_config_api_blueprint, admin_user_api_blueprint]
+admin_apis : list[Blueprint] = [admin_auth_api_blueprint, admin_config_api_blueprint, admin_user_api_blueprint]

app/api/v1/admin/schema/adminAuth.py

@@ -0,0 +1,26 @@
+"""
+Schema for admin authentication API
+"""
+
+from marshmallow import Schema, fields
+
+
+class AdminAuthBasicPostSchema(Schema):
+    """
+    Data schema for admin login
+    """
+    username = fields.String(required=True)
+    password = fields.String(required=True)
+
+    @classmethod
+    def example(cls) -> dict:
+        """
+        Example for the POST admin login
+
+        :return: Example data dict
+        :rtype: dict
+        """
+        return {
+            "username": "admin",
+            "password": "admin"
+        }

app/auth/Admin.py

@@ -0,0 +1,34 @@
+"""
+Admin user representation for authentication
+"""
+
+
+class Admin:
+    """
+    Represents an authenticated admin
+    """
+
+    def __init__(self, uid: str = "") -> None:
+        """
+        Initialize an Admin instance
+
+        :param uid: Admin username/UID
+        :type uid: str
+        """
+        self.uid = uid
+        self.authenticated = True
+        self.anonymous = False
+
+    def __repr__(self) -> str:
+        return f"{self.__class__.__name__}(uid='{self.uid}')"
+
+
+class AdminAnonymous(Admin):
+    """
+    Represents a non-authenticated admin (anonymous)
+    """
+
+    def __init__(self) -> None:
+        super().__init__(uid="anonymous")
+        self.authenticated = False
+        self.anonymous = True