Add api.config module for Settings

.github/workflows/main.yml

@@ -50,11 +50,17 @@ jobs:
           pylint api.main
           pylint api.models
           pylint api.pubsub
+          pylint api.user_manager
+          pylint api.user_models
           pylint tests/unit_tests
 
       - name: Export environment variables
         run: |
           echo "SECRET_KEY=$(openssl rand -hex 32)" >> $GITHUB_ENV
+          echo "SMTP_HOST=smtp.gmail.com" >> $GITHUB_ENV
+          echo "SMTP_PORT=465" >> $GITHUB_ENV
+          echo "[email protected]" >> $GITHUB_ENV
+          echo "EMAIL_PASSWORD=random" >> $GITHUB_ENV
 
       - name: Run pytest
         run: |

.github/workflows/test.yml

@@ -30,6 +30,10 @@ jobs:
       - name: Export environment variables
         run: |
           echo "SECRET_KEY=$(openssl rand -hex 32)" > .env
+          echo "SMTP_HOST=smtp.gmail.com" >> .env
+          echo "SMTP_PORT=465" >> .env
+          echo "[email protected]" >> .env
+          echo "EMAIL_PASSWORD=random" >> .env
 
       - name: Build docker images
         run: docker-compose -f test-docker-compose.yaml build

api/admin.py

@@ -17,10 +17,11 @@
 
 from .auth import Authentication
 from .db import Database
-from .models import User, UserGroup, UserProfile
+from .models import UserGroup
+from .user_models import User
 
 
 async def setup_admin_group(db, admin_group):
     group_obj = await db.find_one(UserGroup, name=admin_group)
     if group_obj is None:
         print(f"Creating {admin_group} group...")
@@ -28,7 +29,7 @@
     return group_obj
 
 
 async def setup_admin_user(db, username, email, admin_group):
     user_obj = await db.find_one_by_attributes(User,
                                                {'profile.username': username})
     if user_obj:
@@ -42,21 +43,20 @@
         return None
     hashed_password = Authentication.get_password_hash(password)
     print(f"Creating {username} user...")
-    profile = UserProfile(
+    return await db.create(User(
         username=username,
         hashed_password=hashed_password,
         email=email,
-        groups=[admin_group]
-    )
-    return await db.create(User(
-        profile=profile
+        groups=[admin_group],
+        is_superuser=1
     ))
 
 
 async def main(args):
     db = Database(args.mongo, args.database)
+    await db.initialize_beanie()
     group = await setup_admin_group(db, args.admin_group)
     user = await setup_admin_user(db, args.username, args.email, group)
     return True
 
 

api/auth.py

@@ -1,127 +1,54 @@
 # SPDX-License-Identifier: LGPL-2.1-or-later
 #
-# Copyright (C) 2021 Collabora Limited
+# Copyright (C) 2021-2023 Collabora Limited
 # Author: Guillaume Tucker <[email protected]>
+# Author: Jeny Sadadia <[email protected]>
 
 """User authentication utilities"""
 
-from datetime import datetime, timedelta
-from fastapi.security import OAuth2PasswordBearer
-from jose import JWTError, jwt
 from passlib.context import CryptContext
-from pydantic import BaseModel, BaseSettings, Field
-from .db import Database
-from .models import User
-
-
-class Token(BaseModel):
-    """Authentication token model"""
-    access_token: str = Field(
-        description='Authentication access token'
-    )
-    token_type: str = Field(
-        description='Access token type e.g. Bearer'
-    )
-
-
-class Settings(BaseSettings):
-    """Authentication settings"""
-    secret_key: str
-    algorithm: str = "HS256"
-    # Set to None so tokens don't expire
-    access_token_expire_minutes: float = None
+from fastapi_users.authentication import (
+    AuthenticationBackend,
+    BearerTransport,
+    JWTStrategy,
+)
+from .config import AuthSettings
 
 
 class Authentication:
-    """Authentication utility class
-
-    This class accepts a single argument `database` in its constructor, which
-    should be a db.Database object.
-    """
+    """Authentication utility class"""
 
     CRYPT_CTX = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
-    def __init__(self, database: Database, token_url: str, user_scopes: dict):
-        self._db = database
-        self._settings = Settings()
-        self._user_scopes = user_scopes
-        self._oauth2_scheme = OAuth2PasswordBearer(
-                tokenUrl=token_url,
-                scopes=self._user_scopes
-        )
-
-    @property
-    def oauth2_scheme(self):
-        """Get authentication scheme"""
-        return self._oauth2_scheme
+    def __init__(self, token_url: str):
+        self._settings = AuthSettings()
+        self._token_url = token_url
 
     @classmethod
     def get_password_hash(cls, password):
         """Get a password hash for a given clear text password string"""
         return cls.CRYPT_CTX.hash(password)
 
-    @classmethod
-    def verify_password(cls, password_hash, user):
-        """Verify that the password hash matches the user's password"""
-        return cls.CRYPT_CTX.verify(password_hash, user.hashed_password)
+    def get_jwt_strategy(self) -> JWTStrategy:
+        """Get JWT strategy for authentication backend"""
+        return JWTStrategy(
+            secret=self._settings.secret_key,
+            algorithm=self._settings.algorithm,
+            lifetime_seconds=self._settings.access_token_expire_seconds
+        )
 
-    async def authenticate_user(self, username: str, password: str):
-        """Authenticate a username / password pair
+    def get_user_authentication_backend(self):
+        """Authentication backend for user management
 
-        Look up a `User` in the database with the provided `username`
-        and check whether the provided clear text `password` matches the hash
-        associated with it.
+        Authentication backend for `fastapi-users` is composed of two
+        parts: Transaport and Strategy.
+        Transport is a mechanism for token transmisson i.e. bearer or cookie.
+        Strategy is a method to generate and secure tokens. It can be JWT,
+        database or Redis.
         """
-        user = await self._db.find_one_by_attributes(
-            User, {'profile.username': username})
-        if not user:
-            return False
-        if not self.verify_password(password, user.profile):
-            return False
-        return user.profile
-
-    def create_access_token(self, data: dict):
-        """Create a JWT access token using the provided arbitrary `data`"""
-        to_encode = data.copy()
-        if self._settings.access_token_expire_minutes:
-            expires_delta = timedelta(
-                    minutes=self._settings.access_token_expire_minutes
-                    )
-            expire = datetime.utcnow() + expires_delta
-            to_encode.update({"exp": expire})
-        encoded_jwt = jwt.encode(
-            to_encode,
-            self._settings.secret_key, algorithm=self._settings.algorithm
-            )
-        return encoded_jwt
-
-    async def get_current_user(self, token, security_scopes):
-        """Decode the given JWT `token` and look up a matching `User`"""
-        try:
-            payload = jwt.decode(
-                token,
-                self._settings.secret_key,
-                algorithms=[self._settings.algorithm]
-            )
-            username: str = payload.get("sub")
-            token_scopes = payload.get("scopes", [])
-            if username is None:
-                return None, "Could not validate credentials"
-
-            for scope in security_scopes:
-                if scope not in token_scopes:
-                    return None, "Access denied"
-
-        except JWTError as error:
-            return None, str(error)
-
-        user = await self._db.find_one_by_attributes(
-            User, {'profile.username': username})
-        return user, None
-
-    async def validate_scopes(self, requested_scopes):
-        """Check if requested scopes are valid user scopes"""
-        for scope in requested_scopes:
-            if scope not in self._user_scopes:
-                return False, scope
-        return True, None
+        bearer_transport = BearerTransport(tokenUrl=self._token_url)
+        return AuthenticationBackend(
+            name="jwt",
+            transport=bearer_transport,
+            get_strategy=self.get_jwt_strategy,
+        )

api/config.py

@@ -0,0 +1,32 @@
+# SPDX-License-Identifier: LGPL-2.1-or-later
+#
+# Copyright (C) 2023 Collabora Limited
+# Author: Jeny Sadadia <[email protected]>
+
+"""Module settings"""
+
+from pydantic import BaseSettings, EmailStr
+
+
+class AuthSettings(BaseSettings):
+    """Authentication settings"""
+    secret_key: str
+    algorithm: str = "HS256"
+    # Set to None so tokens don't expire
+    access_token_expire_seconds: float = None
+
+
+class PubSubSettings(BaseSettings):
+    """Pub/Sub settings loaded from the environment"""
+    cloud_events_source: str = "https://api.kernelci.org/"
+    redis_host: str = "redis"
+    redis_db_number: int = 1
+    keep_alive_period: int = 45
+
+
+class EmailSettings(BaseSettings):
+    """Email settings"""
+    smtp_host: str
+    smtp_port: int
+    email_sender: EmailStr
+    email_password: str

api/db.py

@@ -7,9 +7,11 @@
 """Database abstraction"""
 
 from bson import ObjectId
+from beanie import init_beanie
 from fastapi_pagination.ext.motor import paginate
 from motor import motor_asyncio
-from .models import Hierarchy, Node, User, Regression, UserGroup
+from .models import Hierarchy, Node, Regression, UserGroup
+from .user_models import User
 
 
 class Database:
@@ -39,6 +41,15 @@ def __init__(self, service='mongodb://db:27017', db_name='kernelci'):
         self._motor = motor_asyncio.AsyncIOMotorClient(service)
         self._db = self._motor[db_name]
 
+    async def initialize_beanie(self):
+        """Initialize Beanie ODM to use `fastapi-users` tools for MongoDB"""
+        await init_beanie(
+            database=self._db,
+            document_models=[
+                User,
+            ],
+        )
+
     def _get_collection(self, model):
         col = self.COLLECTIONS[model]
         return self._db[col]

api/email_sender.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+#
+# SPDX-License-Identifier: LGPL-2.1-or-later
+#
+# Copyright (C) 2023 Jeny Sadadia
+# Author: Jeny Sadadia <[email protected]>
+
+"""SMTP Email Sender module"""
+
+from email.mime.multipart import MIMEMultipart
+import email
+import email.mime.text
+import smtplib
+from .config import EmailSettings
+
+
+class EmailSender:
+    """Class to send email report using SMTP"""
+    def __init__(self):
+        self._settings = EmailSettings()
+
+    def _smtp_connect(self):
+        """Method to create a connection with SMTP server"""
+        if self._settings.smtp_port == 465:
+            smtp = smtplib.SMTP_SSL(self._settings.smtp_host,
+                                    self._settings.smtp_port)
+        else:
+            smtp = smtplib.SMTP(self._settings.smtp_host,
+                                self._settings.smtp_port)
+            smtp.starttls()
+        smtp.login(self._settings.email_sender,
+                   self._settings.email_password)
+        return smtp
+
+    def _create_email(self, email_subject, email_content, email_recipient):
+        """Method to create an email message from email subject, contect,
+        sender, and receiver"""
+        email_msg = MIMEMultipart()
+        email_text = email.mime.text.MIMEText(email_content, "plain", "utf-8")
+        email_text.replace_header('Content-Transfer-Encoding', 'quopri')
+        email_text.set_payload(email_content, 'utf-8')
+        email_msg.attach(email_text)
+        email_msg['To'] = email_recipient
+        email_msg['From'] = self._settings.email_sender
+        email_msg['Subject'] = email_subject
+        return email_msg
+
+    def _send_email(self, email_msg):
+        """Method to send an email message using SMTP"""
+        smtp = self._smtp_connect()
+        if smtp:
+            smtp.send_message(email_msg)
+            smtp.quit()
+
+    def create_and_send_email(self, email_subject, email_content,
+                              email_recipient):
+        """Method to create and send email"""
+        email_msg = self._create_email(
+            email_subject, email_content, email_recipient
+        )
+        self._send_email(email_msg)