[v2] Add CAS auth

.pre-commit-config.yaml

@@ -6,7 +6,7 @@ repos:
       hooks:
       - id: black
         language: python
-        types: [python]
+        types: [python3]
         args: ["--line-length=120"]
 
   -   repo: https://github.com/PyCQA/autoflake

README.md

@@ -206,6 +206,10 @@ To see the bot logs when running with docker in detached mode (`-d`), use the [d
 
 Once the ULB role is set, when a new user joins the server, either they are already registered (from another of your servers) in which case they will get the `@ULB` role and get renamed, or they are not registered yet and will receive a DM message with the instructions to register themselves using the `/ulb` command.
 
+* `/feedback`
+
+Send a feedback directly from discord.
+
 ### Admin server
 
 * `/user add`
@@ -224,6 +228,14 @@ Edit info of a user.
 
 Delete a user.
 
+* `/server info`
+
+Get information about a guild (ULB role, number of registered members, ...)
+
+* `/stats`
+
+Get statistics about the bot usage (nombre of configured servers, number of registered users, ...)
+
 * `/update`
 
 This forces a total update of the database and of all the servers. Since the bot already does this automatically at startup and after each reconnection, the only normal usecase for this would be if you manually add an entry (server or user) to the google sheet instead of using the `/user add` command above, we don't recommend manually editing the google sheet.

bot/bot.py

@@ -96,11 +96,24 @@ def load_commands(self) -> None:
                     )
                     self.cog_not_loaded.append(extension)
 
-    async def send_error_log(self, interaction: ApplicationCommandInteraction, error: Exception):
+    async def send_error_log(self, tb: str):
+
+        n = len(tb) // 4050
+
+        #Logs need to be diveded into multiple embed due to size limitation
+        # TODO Check if we can use a list of embeds and one message
+        # TODO Make it dynamic base on the message size from the lib (check library version, maybe need to upgrade)
+        for i in range(n):
+            await self.log_channel.send(embed=disnake.Embed(description=f"```python\n{tb[4050*i:4050*(i+1)]}```"))
+        await self.log_channel.send(embed=disnake.Embed(description=f"```python\n{tb[4050*n:]}```"))
+
+    async def send_cmd_error_log(self, interaction: ApplicationCommandInteraction, error: Exception):
         tb = self.tracebackEx(error)
         logging.error(
             f"{error} raised on command /{interaction.application_command.name} from {interaction.guild.name+'#'+interaction.channel.name if interaction.guild else 'DM'} by {interaction.author.name}.\n{tb}"
         )
+
+        #Send error msg to the user
         await interaction.send(
             content=self.owner.mention,
             embed=disnake.Embed(
@@ -110,17 +123,16 @@ async def send_error_log(self, interaction: ApplicationCommandInteraction, error
             ),
             delete_after=10,
         )
+
+        #Send logs to admins
         await self.log_channel.send(
             embed=disnake.Embed(title=f":x: __** ERROR**__ :x:", description=f"```{error}```").add_field(
                 name=f"Raised on command :",
                 value=f"**/{interaction.application_command.name}:{interaction.id}** from {interaction.guild.name+'#'+interaction.channel.name if interaction.guild else 'DM'} by {interaction.author.mention} at {interaction.created_at} with options\n```{interaction.filled_options}```"
                 + (f" and target\n``'{interaction.target}``'." if interaction.target else "."),
             )
         )
-        n = len(tb) // 4050
-        for i in range(n):
-            await self.log_channel.send(embed=disnake.Embed(description=f"```python\n{tb[4050*i:4050*(i+1)]}```"))
-        await self.log_channel.send(embed=disnake.Embed(description=f"```python\n{tb[4050*n:]}```"))
+        await self.send_error_log(tb)
 
     async def on_slash_command(self, interaction: disnake.ApplicationCommandInteraction) -> None:
         logging.trace(
@@ -138,13 +150,13 @@ async def on_message_command(self, interaction: disnake.MessageCommandInteractio
         )
 
     async def on_slash_command_error(self, interaction: ApplicationCommandInteraction, error: Exception) -> None:
-        await self.send_error_log(interaction, error)
+        await self.send_cmd_error_log(interaction, error)
 
     async def on_user_command_error(self, interaction: disnake.UserCommandInteraction, error: Exception) -> None:
-        await self.send_error_log(interaction, error)
+        await self.send_cmd_error_log(interaction, error)
 
     async def on_message_command_error(self, interaction: disnake.MessageCommandInteraction, error: Exception) -> None:
-        await self.send_error_log(interaction, error)
+        await self.send_cmd_error_log(interaction, error)
 
     async def on_slash_command_completion(self, interaction: disnake.ApplicationCommandInteraction) -> None:
         logging.trace(

classes/database.py

@@ -94,7 +94,7 @@ def loaded(cls) -> bool:
         return cls._loaded
 
     @classmethod
-    def load(cls, bot: Bot) -> None:
+    async def load(cls, bot: Bot) -> bool:
         """Load the data from the google sheet.
 
         Returns
@@ -104,31 +104,35 @@ def load(cls, bot: Bot) -> None:
             - Guild: `Dict[disnake.Guild, disnake.Role]`
             - Users: `Dict[disnake.User, UlbUser]]`
         """
-        # First time this is call, we need to load the credentials and the sheet
-        if not cls._sheet:
-            cred_dict = {}
-            cred_dict["type"] = os.getenv("GS_TYPE")
-            cred_dict["project_id"] = os.getenv("GS_PROJECT_ID")
-            cred_dict["auth_uri"] = os.getenv("GS_AUTHOR_URI")
-            cred_dict["token_uri"] = os.getenv("GS_TOKEN_URI")
-            cred_dict["auth_provider_x509_cert_url"] = os.getenv("GS_AUTH_PROV")
-            cred_dict["client_x509_cert_url"] = os.getenv("GS_CLIENT_CERT_URL")
-            cred_dict["private_key"] = os.getenv("GS_PRIVATE_KEY").replace(
-                "\\n", "\n"
-            )  # Python add a '\' before any '\n' when loading a str
-            cred_dict["private_key_id"] = os.getenv("GS_PRIVATE_KEY_ID")
-            cred_dict["client_email"] = os.getenv("GS_CLIENT_EMAIL")
-            cred_dict["client_id"] = int(os.getenv("GS_CLIENT_ID"))
-            creds = ServiceAccountCredentials.from_json_keyfile_dict(cred_dict, cls._scope)
-            cls._client = gspread.authorize(creds)
-            logging.info("[Database] Google sheet credentials loaded.")
-
-            # Open google sheet
-            cls._sheet = cls._client.open_by_url(os.getenv("GOOGLE_SHEET_URL"))
-            cls._users_ws = cls._sheet.worksheet("users")
-            cls._guilds_ws = cls._sheet.worksheet("guilds")
-
-            logging.info("[Database] Spreadsheed loaded")
+        try:
+            # First time this is call, we need to load the credentials and the sheet
+            if not cls._sheet:
+                cred_dict = {}
+                cred_dict["type"] = os.getenv("GS_TYPE")
+                cred_dict["project_id"] = os.getenv("GS_PROJECT_ID")
+                cred_dict["auth_uri"] = os.getenv("GS_AUTHOR_URI")
+                cred_dict["token_uri"] = os.getenv("GS_TOKEN_URI")
+                cred_dict["auth_provider_x509_cert_url"] = os.getenv("GS_AUTH_PROV")
+                cred_dict["client_x509_cert_url"] = os.getenv("GS_CLIENT_CERT_URL")
+                cred_dict["private_key"] = os.getenv("GS_PRIVATE_KEY").replace(
+                    "\\n", "\n"
+                )  # Python add a '\' before any '\n' when loading a str
+                cred_dict["private_key_id"] = os.getenv("GS_PRIVATE_KEY_ID")
+                cred_dict["client_email"] = os.getenv("GS_CLIENT_EMAIL")
+                cred_dict["client_id"] = int(os.getenv("GS_CLIENT_ID"))
+                creds = ServiceAccountCredentials.from_json_keyfile_dict(cred_dict, cls._scope)
+                cls._client = gspread.authorize(creds)
+                logging.info("[Database] Google sheet credentials loaded.")
+
+                # Open google sheet
+                cls._sheet = cls._client.open_by_url(os.getenv("GOOGLE_SHEET_URL"))
+                cls._users_ws = cls._sheet.worksheet("users")
+                cls._guilds_ws = cls._sheet.worksheet("guilds")
+
+                logging.info("[Database] Spreadsheed loaded")
+        except (ValueError, gspread.exceptions.SpreadsheetNotFound, gspread.exceptions.WorksheetNotFound) as err:
+            await bot.send_error_log(bot.tracebackEx(err))
+            return
 
         logging.info("[Database] Loading data...")
 

classes/user.py

@@ -0,0 +1,35 @@
+# -*- coding: utf-8 -*-
+import asyncio
+import logging
+import os
+from typing import Dict
+from typing import Optional
+#from flask_login import 
+
+# TODO: compare with the database.py UlbUser class ?
+
+class User: 
+#class User(flask_login.UserMixin):
+    """Represent an UlbUser
+
+    Parameters
+    ----------
+    name: `str`
+        The full name of the user
+    email: `str`
+        The ulb email address of the user
+    site_lang: `str`
+        The website language preferred of the user (default: 'fr')
+    """
+    def __init__(self, ulbid: str, name: str, email: str):
+        self.ulbid: str = ulbid
+        self.name: str = name
+        self.email: str = email
+        self.site_lang: str = "fr"
+        self.is_authenticated = False # for flask_login
+        self.is_active = True # for flask_login
+        self.is_anonymous = False
+
+    def get_id() -> str:
+        # for flask_login
+        return self.ulbid

cogs/Admin.py

@@ -28,11 +28,12 @@ def __init__(self, bot: Bot):
     )
     async def update(self, inter: disnake.ApplicationCommandInteraction):
         await inter.response.defer(ephemeral=True)
-        Database.load(self.bot)
-        await utils.update_all_guilds()
-        await inter.edit_original_response(
-            embed=disnake.Embed(description="All servers updated !", color=disnake.Color.green())
-        )
+        await Database.load(self.bot)
+        if (Database.loaded):
+            await utils.update_all_guilds()
+            await inter.edit_original_response(
+                embed=disnake.Embed(description="All servers updated !", color=disnake.Color.green())
+            )
 
     @commands.slash_command(
         name="yearly-update",

cogs/Ulb.py

@@ -19,10 +19,11 @@ def __init__(self, bot: Bot):
 
     @commands.Cog.listener("on_ready")
     async def on_ready(self):
-        Database.load(self.bot)
-        Registration.setup(self)
-        logging.info("[Cog:Ulb] Ready !")
-        await utils.update_all_guilds()
+        await Database.load(self.bot)
+        if (Database.loaded):
+            Registration.setup(self)
+            logging.info("[Cog:Ulb] Ready !")
+            await utils.update_all_guilds()
 
     async def wait_setup(self, inter: disnake.ApplicationCommandInteraction) -> None:
         """Async sleep until GoogleSheet is loaded and RegistrationForm is set"""

docker-compose.yml

@@ -12,3 +12,5 @@ services:
     volumes:
       - ./:/usr/src/ulbdiscordbot
     restart: always # or unless-stopped
+    mem_limit: 1g # memory limit for the container
+    # cpus: 0.5 # cpu limit for the container

main_flask.py

@@ -0,0 +1,121 @@
+from flask import Flask, render_template, request, session, redirect, url_for
+#from flask_login import LoginManager # TODO: user session management
+from user import User # TODO:
+from cas import CASClient # use python-cas : https://djangocas.dev/blog/python-cas-flask-example/
+import os
+
+# TODO: add rate-limiting (Flask-Limiter ?)
+
+app = Flask(__name__)
+app.secret_key = os.getenv("FLASK_SECRET_KEY")
+#login_manager = LoginManager()
+#login_manager.init_app(app)
+
+cas_client = CASClient(
+    version=3, # 2 ?
+    service_url=os.getenv("CAS_SERVICE_URL") # i.e.: service_url='http://localhost:5000/login?next=%2Fprofile',
+    server_url=os.getenv("CAS_SERVER_URL") # i.e.: server_url='https://django-cas-ng-demo-server.herokuapp.com/cas/'
+)
+
+@app.route("/") 
+def home(): 
+    return render_template("home.html")#, user=user)
+
+
+@app.route('/login')
+def login():
+    if "username" in session:
+        # Already logged in
+        if db.get(session["username"]) = None:
+            # Not linked to Discord yet
+            return redirect(url_for("discord_login"))
+        else:
+            return redirect(url_for("user"))
+
+    next = request.args.get("next")
+    ticket = request.args.get("ticket")
+    if not ticket:
+        # No ticket, the request come from end user, send to CAS login
+        cas_login_url = cas_client.get_login_url()
+        app.logger.debug('CAS login URL: %s', cas_login_url)
+        return redirect(cas_login_url)
+
+    # There is a ticket, the request come from CAS as callback.
+    # need call `verify_ticket()` to validate ticket and get user profile.
+    app.logger.debug('ticket: %s', ticket)
+    app.logger.debug('next: %s', next)
+
+    user, attributes, pgtiou = cas_client.verify_ticket(ticket)
+
+    app.logger.debug(
+        'CAS verify ticket response: user: %s, attributes: %s, pgtiou: %s', user, attributes, pgtiou)
+
+    if not user:
+        return 'Failed to verify ticket. Try to login again: <a href="/">Home</a>'
+        #return redirect(url_for(("home"))
+    else:  # Login successfully, redirect according `next` query parameter.
+        session['username'] = user
+        #session["attributes"] = attributes # TODO: check what attributes we get
+        return redirect(next)
+
+@app.route('/logout')
+def logout():
+    redirect_url = url_for('logout_callback', _external=True)
+    cas_logout_url = cas_client.get_logout_url(redirect_url)
+    app.logger.debug('CAS logout URL: %s', cas_logout_url)
+
+    return redirect(cas_logout_url)
+
+@app.route('/logout_callback')
+def logout_callback():
+    # redirect from CAS logout request after CAS logout successfully
+    session.pop('username', None)
+    #return 'Logged out from CAS. <a href="/login">Login</a>'
+    return redirect(url_for(("home")) # ?
+
+@app.route("/discord_login")
+def discord_login(): 
+    if user.site_lang = "en":
+        return render_template("discord-login_en.html"; user=user)
+    else:
+        return render_template("discord-login_fr.html", user=user) 
+
+
+@app.route("/user")
+def user(method=['GET']):
+    if 'username' in session:
+        user = session["username"]
+        #return 'Logged in as %s. <a href="/logout">Logout</a>' % session['username']
+        if user.site_lang = "en":
+            return render_template("user_en.html", user=user)
+        else:
+            return render_template("user_fr.html", user=user)
+    else:
+        return 'Login required. <a href="/">Back to Home</a>', 403
+        #return redirect(url_for("home"))
+
+@app.route("/test") 
+def test():
+    test_user = User(
+        ulbid="test",
+        email="test@example.com"
+    )
+    return render_template("test.html", user=test_user)
+
+@app.route("/ping")
+def ping():
+    return "pong"
+
+
+@app.route("/bot") 
+def bot(): 
+    # TODO: check if bot status is online
+    bot_status = "Online"
+    return render_template("bot.html", status=bot_status) 
+
+
+if __name__ == "__main__":
+    # TODO: run Disnake and Flask in different threads ? asyncio ?
+    app.run(debug=False)
+
+