Merge branch 'main' of git.cs.kau.se:csma/cyber-range/ctfd-plugins into main

README.md

@@ -12,6 +12,7 @@ cp ctfd-plugins/docker-compose.yml CTFd/docker-compose.yml
 cp -r ctfd-plugins/oauth CTFd/CTFd/plugins/
 cp -r ctfd-plugins/init CTFd/CTFd/plugins/
 cp -r ctfd-plugins/crld CTFd/CTFd/plugins/
+cp -r ctfd-plugins/registration CTFd/CTFd/plugins/
 cd CTFd
 docker compose up --build
 ```
@@ -56,6 +57,16 @@ button will reset the instance.
 ### Environment variables
 - CRL_ENABLE # whether the crl plugin is used {true,false}
 
+## registration
+The [registration plugin](registration) adds extra constraints checks to the user 
+registration process.
+
+###  Environment variables
+- REGISTRATION_USERNAME_ALLOWED_REGEX     # allowed characters in username, default allow anything
+- REGISTRATION_USERNAME_MAX_LENGTH        # maximum amount of characters in username, default 128
+- REGISTRATION_USERNAME_MIN_LENGTH        # minimum amount of characters in username, default 1
+- REGISTRATION_USERNAME_CASE_SENSITIVE    # default true due to legacy but needs to be false for CRL
+
 # Todo
 - crl: Communication with CRLd - parse response
 

build-all.sh

@@ -6,5 +6,6 @@ git clone https://github.com/CTFd/CTFd.git
 cp -r oauth CTFd/CTFd/plugins/
 cp -r init CTFd/CTFd/plugins/
 cp -r crl CTFd/CTFd/plugins/
+cp -r registration CTFd/CTFd/plugins/
 docker compose build
 

registration/__init__.py

+import os
+import re
+import requests
+
+from flask import abort
+from flask import render_template
+from flask import redirect, render_template, request, session, url_for
+from flask import current_app as app
+from itsdangerous.exc import BadSignature, BadTimeSignature, SignatureExpired
+from sqlalchemy import func
+
+from CTFd.models import Brackets, UserFieldEntries, UserFields, Users, db
+from CTFd.utils import user as current_user
+from CTFd.utils import config, email, get_config
+from CTFd.utils import validators
+from CTFd.utils.config import is_teams_mode
+from CTFd.utils.decorators import ratelimit
+from CTFd.utils.decorators.visibility import check_registration_visibility
+from CTFd.utils.helpers import get_errors
+from CTFd.utils.logging import log
+from CTFd.utils.security.auth import login_user
+from CTFd.utils.validators import ValidationError
+
+def load(app):
+    @check_registration_visibility
+    @ratelimit(method="POST", limit=10, interval=5)
+    def register_v2():
+        errors = get_errors()
+        if current_user.authed():
+            return redirect(url_for("challenges.listing"))
+
+        num_users_limit = int(get_config("num_users", default=0))
+        num_users = Users.query.filter_by(banned=False, hidden=False).count()
+        if num_users_limit and num_users >= num_users_limit:
+            abort(
+                403,
+                description=f"Reached the maximum number of users ({num_users_limit}).",
+            )
+
+        if request.method == "POST":
+            # registration plugin settings
+            user_allowed_regex = os.getenv("REGISTRATION_USERNAME_ALLOWED_REGEX", None)
+            user_max_len = int(os.getenv("REGISTRATION_USERNAME_MAX_LENGTH", 128))
+            user_min_len = int(os.getenv("REGISTRATION_USERNAME_MIN_LENGTH", 1))
+            user_case_sensitve = os.getenv("REGISTRATION_USERNAME_CASE_SENSITIVE", "true").lower() == "true"
+
+            name = request.form.get("name", "").strip()
+            email_address = request.form.get("email", "").strip().lower()
+            password = request.form.get("password", "").strip()
+
+            website = request.form.get("website")
+            affiliation = request.form.get("affiliation")
+            country = request.form.get("country")
+            registration_code = str(request.form.get("registration_code", ""))
+            bracket_id = request.form.get("bracket_id", None)
+
+            # username checks.
+            if len(name) < user_min_len:
+                errors.append("Pick a longer user name")
+            if len(name) > user_max_len:
+                errors.append("Pick a shorter user name")
+
+            user_query = Users.query.add_columns(Users.name, Users.id)
+            if user_case_sensitve:
+                names = user_query.filter_by(name=name).first()
+            else:
+                names = user_query.filter(func.lower(Users.name) == func.lower(name)).first()
+            if names:
+                errors.append("User name is already taken")
+
+            if user_allowed_regex:
+                allowed = re.compile(user_allowed_regex)
+                if not allowed.match(name):
+                    errors.append("User name contains illegal characters")
+
+            # email checks.
+            emails = (
+                Users.query.add_columns(Users.email, Users.id)
+                .filter_by(email=email_address)
+                .first()
+            )
+            if email.check_email_is_whitelisted(email_address) is False:
+                errors.append("Your email address is not from an allowed domain")
+            if emails:
+                errors.append("That email has already been used")
+            if not validators.validate_email(email_address):
+                errors.append("Please enter a valid email address")
+            if validators.validate_email(name):
+                errors.append("Your user name cannot be an email address")
+
+            # password_checks.
+            if len(password) == 0:
+                errors.append("Pick a longer password")
+            if len(password) > 128:
+                errors.append("Pick a shorter password")
+ 
+            # registration code checks.
+            if get_config("registration_code"):
+                if (
+                    registration_code.lower()
+                    != str(get_config("registration_code", default="")).lower()
+                ):
+                    errors.append("The registration code you entered was incorrect")
+
+            # Process additional user fields
+            fields = {}
+            for field in UserFields.query.all():
+                fields[field.id] = field
+
+            entries = {}
+            for field_id, field in fields.items():
+                value = request.form.get(f"fields[{field_id}]", "").strip()
+                if field.required is True and (value is None or value == ""):
+                    errors.append("Please provide all required fields")
+                    break
+
+                if field.field_type == "boolean":
+                    entries[field_id] = bool(value)
+                else:
+                    entries[field_id] = value
+
+            # country checks.
+            if country:
+                try:
+                    validators.validate_country_code(country)
+                    valid_country = True
+                except ValidationError:
+                    valid_country = False
+            else:
+                valid_country = True
+            if valid_country is False:
+                errors.append("Invalid country")
+
+            # website checks.
+            if website and validators.validate_url(website) is False:
+                errors.append("Websites must be a proper URL starting with http or https")
+
+            # affiliation checks.
+            if affiliation and len(affiliation) < 128:
+                errors.append("Please provide a shorter affiliation")
+
+            # bracket checks.
+            if bracket_id:
+                valid_bracket = bool(
+                    Brackets.query.filter_by(id=bracket_id, type="users").first()
+                )
+            else:
+                if Brackets.query.filter_by(type="users").count():
+                    valid_bracket = False
+                else:
+                    valid_bracket = True
+            if valid_bracket is False:
+                errors.append("Please provide a valid bracket")
+
+            if len(errors) > 0:
+                return render_template(
+                    "register.html",
+                    errors=errors,
+                    name=request.form["name"],
+                    email=request.form["email"],
+                    password=request.form["password"],
+                )
+            else:
+                with app.app_context():
+                    user = Users(
+                        name=name,
+                        email=email_address,
+                        password=password,
+                        bracket_id=bracket_id,
+                    )
+
+                    if website:
+                        user.website = website
+                    if affiliation:
+                        user.affiliation = affiliation
+                    if country:
+                        user.country = country
+
+                    db.session.add(user)
+                    db.session.commit()
+                    db.session.flush()
+
+                    for field_id, value in entries.items():
+                        entry = UserFieldEntries(
+                            field_id=field_id, value=value, user_id=user.id
+                        )
+                        db.session.add(entry)
+                    db.session.commit()
+
+                    login_user(user)
+
+                    if request.args.get("next") and validators.is_safe_url(
+                        request.args.get("next")
+                    ):
+                        return redirect(request.args.get("next"))
+
+                    if config.can_send_mail() and get_config(
+                        "verify_emails"
+                    ):  # Confirming users is enabled and we can send email.
+                        log(
+                            "registrations",
+                            format="[{date}] {ip} - {name} registered (UNCONFIRMED) with {email}",
+                            name=user.name,
+                            email=user.email,
+                        )
+                        email.verify_email_address(user.email)
+                        db.session.close()
+                        return redirect(url_for("auth.confirm"))
+                    else:  # Don't care about confirming users
+                        if (
+                            config.can_send_mail()
+                        ):  # We want to notify the user that they have registered.
+                            email.successful_registration_notification(user.email)
+
+            log(
+                "registrations",
+                format="[{date}] {ip} - {name} registered with {email}",
+                name=user.name,
+                email=user.email,
+            )
+            db.session.close()
+
+            if is_teams_mode():
+                return redirect(url_for("teams.private"))
+
+            return redirect(url_for("challenges.listing"))
+        else:
+            return render_template("register.html", errors=errors)
+
+
+
+    # The format used by the view_functions dictionary is blueprint.view_function_name
+    app.view_functions['auth.register'] = register_v2