Remove Auth class definition entirely

Converted to a series of module methods and properties.

Author: npalaska

lib/pbench/server/api/resources/__init__.py

@@ -13,7 +13,7 @@
 from sqlalchemy.orm.query import Query
 
 from pbench.server import JSON, JSONOBJECT, JSONVALUE, OperationCode, PbenchServerConfig
-from pbench.server.auth.auth import Auth
+import pbench.server.auth.auth as Auth
 from pbench.server.database.models.audit import (
     Audit,
     AuditReason,
@@ -1470,7 +1470,7 @@ def _get_dataset_metadata(
                 native_key = Metadata.get_native_key(i)
                 user_id = None
                 if native_key == Metadata.USER:
-                    user_id = Auth.get_user_id()
+                    user_id = Auth.get_current_user_id()
                 try:
                     metadata[i] = Metadata.getvalue(
                         dataset=dataset, key=i, user_id=user_id

lib/pbench/server/api/resources/datasets_metadata.py

@@ -18,7 +18,7 @@
     ParamType,
     Schema,
 )
-from pbench.server.auth.auth import Auth
+import pbench.server.auth.auth as Auth
 from pbench.server.database.models.audit import AuditType
 from pbench.server.database.models.datasets import (
     Metadata,
@@ -191,7 +191,7 @@ def _put(
             native_key = Metadata.get_native_key(k)
             user_id = None
             if native_key == Metadata.USER:
-                user_id = Auth.get_user_id()
+                user_id = Auth.get_current_user_id()
             try:
                 Metadata.setvalue(key=k, value=v, dataset=dataset, user_id=user_id)
             except MetadataError as e:

lib/pbench/server/api/resources/query_apis/__init__.py

@@ -28,7 +28,7 @@
     SchemaError,
     UnauthorizedAccess,
 )
-from pbench.server.auth.auth import Auth
+import pbench.server.auth.auth as Auth
 from pbench.server.database.models.audit import AuditReason, AuditStatus
 from pbench.server.database.models.datasets import Dataset, Metadata
 from pbench.server.database.models.template import Template

lib/pbench/server/api/resources/upload_api.py

@@ -11,7 +11,7 @@
 import humanize
 
 from pbench.common.utils import Cleanup, validate_hostname
-from pbench.server.auth.auth import Auth
+import pbench.server.auth.auth as Auth
 from pbench.server.cache_manager import CacheManager
 from pbench.server.database.models.audit import (
     Audit,

lib/pbench/server/api/resources/users_api.py

@@ -9,7 +9,7 @@
 import jwt
 from sqlalchemy.exc import IntegrityError, SQLAlchemyError
 
-from pbench.server.auth.auth import Auth
+import pbench.server.auth.auth as Auth
 from pbench.server.database.models.active_tokens import ActiveTokens
 from pbench.server.database.models.server_config import ServerConfig
 from pbench.server.database.models.users import User

lib/pbench/server/auth/auth.py

@@ -12,169 +12,168 @@
 from pbench.server.database.models.active_tokens import ActiveTokens
 from pbench.server.database.models.users import User
 
-
-class Auth:
-    token_auth = HTTPTokenAuth("Bearer")
-    secret_key = ""
-
-    @staticmethod
-    def get_user_id() -> Optional[str]:
-        """
-        Returns the user id of the current authenticated user.
-        If the user not authenticated this would return None
-        """
-        user_id = None
-        user = Auth.token_auth.current_user()
-        if user:
-            user_id = str(user.id)
-        return user_id
-
-    @staticmethod
-    def _get_secret_key():
-        if not Auth.secret_key:
-            Auth.secret_key = os.getenv("SECRET_KEY", "my_precious")
-        return Auth.secret_key
-
-    @staticmethod
-    def encode_auth_token(time_delta: datetime.timedelta, user_id: int) -> str:
-        """
-        Generates the Auth Token
-        Args:
-            time_delta: Token lifetime
-            user_id: Authorized user's internal ID
-        Returns:
-            JWT token string
-        """
-        current_utc = datetime.datetime.now(datetime.timezone.utc)
-        payload = {
-            "iat": current_utc,
-            "exp": current_utc + time_delta,
-            "sub": user_id,
-        }
-
-        # Get jwt key
-        jwt_key = Auth._get_secret_key()
-        return jwt.encode(payload, jwt_key, algorithm="HS256")
-
-    @staticmethod
-    def get_auth_token():
-        # get auth token
-        auth_header = request.headers.get("Authorization")
-
-        if not auth_header:
-            abort(
-                HTTPStatus.FORBIDDEN,
-                message="Please add authorization token as 'Authorization: Bearer <session_token>'",
-            )
-
-        try:
-            auth_schema, auth_token = auth_header.split()
-        except ValueError:
+token_auth = HTTPTokenAuth("Bearer")
+_secret_key = ""
+
+
+def get_current_user_id() -> Optional[str]:
+    """Returns the user id of the current authenticated user.
+    If the user is not authenticated returns None
+    """
+    user_id = None
+    user = token_auth.current_user()
+    if user:
+        user_id = str(user.id)
+    return user_id
+
+
+def _get_secret_key():
+    global _secret_key
+    if not _secret_key:
+        _secret_key = os.getenv("SECRET_KEY", "my_precious")
+    return _secret_key
+
+
+def encode_auth_token(time_delta: datetime.timedelta, user_id: int) -> str:
+    """
+    Generates the Auth Token
+    Args:
+        time_delta: Token lifetime
+        user_id: Authorized user's internal ID
+    Returns:
+        JWT token string
+    """
+    current_utc = datetime.datetime.now(datetime.timezone.utc)
+    payload = {
+        "iat": current_utc,
+        "exp": current_utc + time_delta,
+        "sub": user_id,
+    }
+
+    # Get jwt key
+    jwt_key = _get_secret_key()
+    return jwt.encode(payload, jwt_key, algorithm="HS256")
+
+
+def get_auth_token():
+    # get auth token
+    auth_header = request.headers.get("Authorization")
+
+    if not auth_header:
+        abort(
+            HTTPStatus.FORBIDDEN,
+            message="Please add authorization token as 'Authorization: Bearer <session_token>'",
+        )
+
+    try:
+        auth_schema, auth_token = auth_header.split()
+    except ValueError:
+        abort(
+            HTTPStatus.UNAUTHORIZED,
+            message="Malformed Authorization header, please add request header as Authorization: Bearer <session_token>",
+        )
+    else:
+        if auth_schema.lower() != "bearer":
             abort(
                 HTTPStatus.UNAUTHORIZED,
-                message="Malformed Authorization header, please add request header as Authorization: Bearer <session_token>",
+                message="Malformed Authorization header, request needs bearer token: Bearer <session_token>",
             )
-        else:
-            if auth_schema.lower() != "bearer":
-                abort(
-                    HTTPStatus.UNAUTHORIZED,
-                    message="Malformed Authorization header, request needs bearer token: Bearer <session_token>",
-                )
-            return auth_token
-
-    @staticmethod
-    @token_auth.verify_token
-    def verify_auth(auth_token):
-        """
-        Validates the auth token.
-        Note: Since we are not encoding 'aud' claim in our JWT tokens
-        we need to set 'verify_aud' to False while validating the token.
-        With issue https://issues.redhat.com/browse/PBENCH-895 we can
-        set it to True when we start validating third party OIDC tokens.
-        :param auth_token:
-        :return: User object/None
-        """
+        return auth_token
+
+
+@token_auth.verify_token
+def verify_auth(auth_token):
+    """
+    Validates the auth token.
+    Note: Since we are not encoding 'aud' claim in our JWT tokens
+    we need to set 'verify_aud' to False while validating the token.
+    With issue https://issues.redhat.com/browse/PBENCH-895 we can
+    set it to True when we start validating third party OIDC tokens.
+    :param auth_token:
+    :return: User object/None
+    """
+    try:
+        payload = jwt.decode(
+            auth_token,
+            os.getenv("SECRET_KEY", "my_precious"),
+            algorithms="HS256",
+            options={
+                "verify_signature": True,
+                "verify_aud": False,
+                "verify_exp": True,
+            },
+        )
+        user_id = payload["sub"]
+        if ActiveTokens.valid(auth_token):
+            user = User.query(id=user_id)
+            return user
+    except jwt.ExpiredSignatureError:
         try:
-            payload = jwt.decode(
-                auth_token,
-                Auth._get_secret_key(),
-                algorithms="HS256",
-                options={
-                    "verify_signature": True,
-                    "verify_aud": False,
-                    "verify_exp": True,
-                },
-            )
-            user_id = payload["sub"]
-            if ActiveTokens.valid(auth_token):
-                user = User.query(id=user_id)
-                return user
-        except jwt.ExpiredSignatureError:
-            try:
-                ActiveTokens.delete(auth_token)
-            except Exception as e:
-                current_app.logger.error(
-                    "User passed expired token but we could not delete the token from the database. token: {!r}: {}",
-                    auth_token,
-                    e,
-                )
-        except jwt.InvalidTokenError:
-            pass  # Ignore this silently; client is unauthenticated
+            ActiveTokens.delete(auth_token)
         except Exception as e:
-            current_app.logger.exception(
-                "Unexpected exception occurred while verifying the auth token {!r}: {}",
+            current_app.logger.error(
+                "User passed expired token but we could not delete the token from the database. token: {!r}: {}",
                 auth_token,
                 e,
             )
-        return None
-
-    @staticmethod
-    def verify_third_party_token(
-        auth_token: str, oidc_client: OpenIDClient, logger: Logger
-    ) -> bool:
-        """
-        Verify a token provided to the Pbench server which was obtained from a
-        third party identity provider.
-        Args:
-            auth_token: Token to authenticate
-            oidc_client: OIDC client to call to authenticate the token
-        Returns:
-            True if the verification succeeds else False
-        """
-        try:
-            oidc_client.token_introspect_offline(
-                token=auth_token,
-                key=oidc_client.get_oidc_public_key(),
-                audience=oidc_client.client_id,
-                options={
-                    "verify_signature": True,
-                    "verify_aud": True,
-                    "verify_exp": True,
-                },
-            )
-            return True
-        except (
-            jwt.ExpiredSignatureError,
-            jwt.InvalidTokenError,
-            jwt.InvalidAudienceError,
-        ):
-            logger.error("OIDC token verification failed")
-            return False
-        except Exception:
-            logger.exception(
-                "Unexpected exception occurred while verifying the auth token {}",
-                auth_token,
-            )
-
-        if not oidc_client.TOKENINFO_ENDPOINT:
-            logger.warning("Can not perform OIDC online token verification")
-            return False
-
-        try:
-            token_payload = oidc_client.token_introspect_online(
-                token=auth_token, token_info_uri=oidc_client.TOKENINFO_ENDPOINT
-            )
-        except OpenIDClientError:
-            return False
-
-        return "aud" in token_payload and oidc_client.client_id in token_payload["aud"]
+    except jwt.InvalidTokenError:
+        pass  # Ignore this silently; client is unauthenticated
+    except Exception as e:
+        current_app.logger.exception(
+            "Unexpected exception occurred while verifying the auth token {!r}: {}",
+            auth_token,
+            e,
+        )
+    return None
+
+
+def verify_third_party_token(
+    auth_token: str, oidc_client: OpenIDClient, logger: Logger
+) -> bool:
+    """
+    Verify a token provided to the Pbench server which was obtained from a
+    third party identity provider.
+    Args:
+        auth_token: Token to authenticate
+        oidc_client: OIDC client to call to authenticate the token
+    Returns:
+        True if the verification succeeds else False
+    """
+    identity_provider_pubkey = oidc_client.get_oidc_public_key(auth_token)
+    try:
+        oidc_client.token_introspect_offline(
+            token=auth_token,
+            key=identity_provider_pubkey,
+            audience=oidc_client.client_id,
+            options={
+                "verify_signature": True,
+                "verify_aud": True,
+                "verify_exp": True,
+            },
+        )
+        return True
+    except (
+        jwt.ExpiredSignatureError,
+        jwt.InvalidTokenError,
+        jwt.InvalidAudienceError,
+    ):
+        logger.error("OIDC token verification failed")
+        return False
+    except Exception:
+        logger.exception(
+            "Unexpected exception occurred while verifying the auth token {}",
+            auth_token,
+        )
+
+    if not oidc_client.TOKENINFO_ENDPOINT:
+        logger.warning("Can not perform OIDC online token verification")
+        return False
+
+    try:
+        token_payload = oidc_client.token_introspect_online(
+            token=auth_token, token_info_uri=oidc_client.TOKENINFO_ENDPOINT
+        )
+    except OpenIDClientError:
+        return False
+
+    return "aud" in token_payload and oidc_client.client_id in token_payload["aud"]

lib/pbench/test/unit/server/conftest.py

@@ -18,7 +18,7 @@
 from pbench.common.logger import get_pbench_logger
 from pbench.server import PbenchServerConfig
 from pbench.server.api import create_app, get_server_config
-from pbench.server.auth.auth import Auth
+import pbench.server.auth.auth as Auth
 from pbench.server.database.database import Database
 from pbench.server.database.models.active_tokens import ActiveTokens
 from pbench.server.database.models.datasets import Dataset, Metadata, States