[WIP] Make anitya use authlib instead of social_auth

Containerfile.dev

@@ -37,8 +37,4 @@ RUN pushd anitya/static && npm install && popd
 RUN poetry build
 RUN pip install dist/*.whl
 
-# Hotfix for social_auth-sqlalchemy
-# Should be removed when we switch to something else
-RUN sed -i 's/base64.encodestring/base64.encodebytes/g' /usr/local/lib/python3.12/site-packages/social_sqlalchemy/storage.py
-
 CMD ["sh","-c", "poetry build && pip install dist/*.whl && eval '$START_COMMAND'"]

anitya/admin.py

@@ -6,7 +6,6 @@
 
 import flask
 from dateutil import parser
-from social_flask_sqlalchemy import models as social_models
 from sqlalchemy.orm.exc import NoResultFound
 
 import anitya
@@ -678,13 +677,6 @@ def delete_user(user_id):
 
     if form.validate_on_submit():
         if confirm:
-            social_auth_records = (
-                Session.query(social_models.UserSocialAuth)
-                .filter_by(user_id=user_id)
-                .all()
-            )
-            for record in social_auth_records:
-                Session.delete(record)
             Session.delete(user)
             Session.commit()
             flask.flash(f"User {user_name} has been removed", "success")

anitya/app.py

@@ -15,25 +15,20 @@
 import logging.handlers
 
 import flask
+from authlib.integrations.flask_client import OAuth
 from flask_login import LoginManager, current_user, user_logged_in
-from social_core.backends.utils import load_backends
-from social_core.exceptions import AuthException
-from social_flask.routes import social_auth
-from social_flask_sqlalchemy import models as social_models
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm.exc import NoResultFound
 
 import anitya.lib
 import anitya.mail_logging
-from anitya import __version__
+from anitya import __version__, admin, api, api_v2, auth, authentication, ui
 from anitya.config import config as anitya_config
 from anitya.db import Session
 from anitya.db import initialize as initialize_db
 from anitya.db import models
 from anitya.lib import utilities
 
-from . import admin, api, api_v2, authentication, ui
-
 
 def create(config=None):
     """
@@ -54,19 +49,6 @@ def create(config=None):
     app.config.update(config)
     initialize_db(config)
 
-    app.register_blueprint(social_auth)
-    if len(social_models.UserSocialAuth.__table_args__) == 0:
-        # This is a bit of a hack - this initialization call sets up the SQLAlchemy
-        # models with our own models and multiple calls to this function will cause
-        # SQLAlchemy to fail with sqlalchemy.exc.InvalidRequestError. Only calling it
-        # when there are no table arguments should ensure we only call it one time.
-        #
-        # Be aware that altering the configuration values this function uses, namely
-        # the SOCIAL_AUTH_USER_MODEL, after the first time ``create`` has been called
-        # will *not* cause the new configuration to be used for subsequent calls to
-        # ``create``.
-        social_models.init_social(app, Session)
-
     login_manager = LoginManager()
     login_manager.user_loader(authentication.load_user_from_session)
     login_manager.request_loader(authentication.load_user_from_request)
@@ -91,16 +73,20 @@ def create(config=None):
     app.register_blueprint(ui.ui_blueprint)
     app.register_blueprint(api.api_blueprint)
 
+    oauth = OAuth(app)
+    for auth_backend in app.config.get("AUTHLIB_ENABLED_BACKENDS", []):
+        oauth.register(auth_backend.lower())
+
+    app.register_blueprint(auth.create_oauth_blueprint(oauth))
+
     app.before_request(global_user)
     app.teardown_request(shutdown_session)
     app.register_error_handler(IntegrityError, integrity_error_handler)
-    app.register_error_handler(AuthException, auth_error_handler)
+    # TODO: Need to change for authlib
+    # app.register_error_handler(AuthException, auth_error_handler)
 
     app.context_processor(inject_variable)
 
-    # subscribe to signals
-    user_logged_in.connect(when_user_log_in, app)
-
     if app.config.get("EMAIL_ERRORS"):
         # If email logging is configured, set up the anitya logger with an email
         # handler for any ERROR-level logs.
@@ -141,9 +127,7 @@ def inject_variable():
         justedit=justedit,
         cron_status=cron_status,
         user=current_user,
-        available_backends=load_backends(
-            anitya_config["SOCIAL_AUTH_AUTHENTICATION_BACKENDS"]
-        ),
+        available_backends=anitya_config["AUTHLIB_ENABLED_BACKENDS"],
     )
 
 
@@ -157,71 +141,4 @@ def integrity_error_handler(error):
     Returns:
         tuple: A tuple of (message, HTTP error code).
     """
-    # Because social auth provides the route and raises the exception, this is
-    # the simplest way to turn the error into a nicely formatted error message
-    # for the user.
-    if "email" in error.params:
-        Session.rollback()
-        other_user = models.User.query.filter_by(email=error.params["email"]).one()
-        try:
-            social_auth_user = other_user.social_auth.filter_by(
-                user_id=other_user.id
-            ).one()
-            msg = (
-                "Error: There's already an account associated with your email, "
-                f"authenticate with {social_auth_user.provider}."
-            )
-            return msg, 400
-        # This error happens only if there is account without provider info
-        except NoResultFound:
-            Session.delete(other_user)
-            Session.commit()
-            msg = (
-                "Error: There was already an existing account with missing provider. "
-                "So we removed it. "
-                "Please try to log in again."
-            )
-            return msg, 500
-
     return "The server encountered an unexpected error", 500
-
-
-def auth_error_handler(error):
-    """
-    Flask error handler for unhandled AuthException errors.
-
-    Args:
-        error (AuthException): The exception to be handled.
-
-    Returns:
-        tuple: A tuple of (message, HTTP error code).
-    """
-    # Because social auth openId backend provides route and raises the exceptions,
-    # this is the simplest way to turn error into nicely formatted error message.
-    msg = (
-        f"Error: There was an error during authentication '{error}', "
-        "please check the provided url."
-    )
-    return msg, 400
-
-
-def when_user_log_in(sender, user):
-    """
-    This catches the signal when user is logged in.
-    It checks if the user has associated entry in user_social_auth.
-
-    Args:
-        sender (flask.Flask): Current app object that emitted signal.
-            Not used by this method.
-        user (models.User): User that is logging in.
-
-    Raises:
-        sqlalchemy.exc.IntegrityError: When user_social_auth table entry is
-        missing.
-    """
-    if user.social_auth.count() == 0:
-        raise IntegrityError(
-            "Missing social_auth table",
-            {"social_auth": None, "email": user.email},
-            None,
-        )

anitya/auth.py

@@ -0,0 +1,59 @@
+import flask
+import flask_login
+
+from anitya.db import Session, User
+
+
+def create_oauth_blueprint(oauth):
+    oauth_blueprint = flask.Blueprint("oauth", __name__)
+
+    @oauth_blueprint.route("/oauthlogin/<name>")
+    def oauthlogin(name):
+        client = oauth.create_client(name)
+        if client is None:
+            flask.abort(400)
+        redirect_uri = flask.url_for(".auth", name=name, _external=True)
+        return client.authorize_redirect(redirect_uri)
+
+    @oauth_blueprint.route("/auth/<name>")
+    def auth(name):
+        client = oauth.create_client(name)
+        if client is None:
+            flask.abort(400)
+        id_token = flask.request.values.get("id_token")
+        if flask.request.values.get("code"):
+            token = client.authorize_access_token()
+            if id_token:
+                token["id_token"] = id_token
+        elif id_token:
+            token = {"id_token": id_token}
+
+        # for Google
+        if "id_token" in token:
+            user_info = client.parse_id_token(token)
+        # for Github
+        else:
+            client.response = client.get("user", token=token)
+            user_info = client.response.json()
+            if user_info["email"] is None:
+                resp = client.get("user/emails", token=token)
+                resp.raise_for_status()
+                data = resp.json()
+                user_info["email"] = next(
+                    email["email"] for email in data if email["primary"]
+                )
+
+        # Check if the user exists
+        user = User.query.filter(User.email == user_info["email"]).first()
+        if not user:
+            # TODO: Should create the user (and openid connections) if it does not exist
+            new_user = User(email=user_info["email"], username=user_info["email"])
+            Session.add(new_user)
+            Session.commit()
+            user = new_user
+        flask_login.login_user(user)
+
+        # TODO: Process next not to just redirect with the main page
+        return flask.redirect("/")
+
+    return oauth_blueprint

anitya/authentication.py

@@ -22,13 +22,12 @@
 Anitya uses `Flask-Login`_ for user session management. It handles logging in,
 logging out, and remembering users’ sessions over extended periods of time.
 
-In addition, Anitya uses `Python Social Auth`_ to authenticate users from various
+In addition, Anitya uses `Authlib`_ to authenticate users from various
 third-party identity providers. It handles logging the user in and creating
 :class:`anitya.db.models.User` objects as necessary.
 
 .. _Flask-Login: https://flask-login.readthedocs.io/en/latest/
-.. _Python Social Auth:
-    https://python-social-auth.readthedocs.io/en/latest/
+.. _Authlib: https://docs.authlib.org/en/latest/
 """
 import logging
 import uuid

anitya/config.py

@@ -65,20 +65,6 @@
     EMAIL_ERRORS=False,
     BLACKLISTED_USERS=[],
     SESSION_PROTECTION="strong",
-    SOCIAL_AUTH_AUTHENTICATION_BACKENDS=(
-        "social_core.backends.fedora.FedoraOpenId",
-        "social_core.backends.gitlab.GitLabOAuth2",
-        "social_core.backends.github.GithubOAuth2",
-        "social_core.backends.google.GoogleOAuth2",
-        "social_core.backends.open_id.OpenIdAuth",
-    ),
-    SOCIAL_AUTH_STORAGE="social_flask_sqlalchemy.models.FlaskStorage",
-    SOCIAL_AUTH_USER_MODEL="anitya.db.models.User",
-    # Force the application to require HTTPS on authentication redirects.
-    SOCIAL_AUTH_REDIRECT_IS_HTTPS=True,
-    SOCIAL_AUTH_LOGIN_URL="/login/",
-    SOCIAL_AUTH_LOGIN_REDIRECT_URL="/",
-    SOCIAL_AUTH_LOGIN_ERROR_URL="/login-error/",
     DEFAULT_REGEX=r"(?i)%(name)s(?:[-_]?(?:minsrc|src|source))?[-_]([^-/_\s]+?(?:[-_]"
     r"(?:rc|devel|dev|alpha|beta)\d+)?)(?:[-_](?:minsrc|src|source|asc|release))?"
     r"\.(?:tar|t[bglx]z|tbz2|zip)",
@@ -90,6 +76,17 @@
     # project will be automatically removed, if no version was retrieved yet
     CHECK_ERROR_THRESHOLD=100,
     DISTRO_MAPPING_LINKS={},
+    # Enabled authentication backends
+    AUTHLIB_ENABLED_BACKENDS=["Fedora", "GitHub", "Google"],
+    # Token for GitHub API
+    GITHUB_ACCESS_TOKEN_URL="https://github.com/login/oauth/access_token",
+    GITHUB_AUTHORIZE_URL="https://github.com/login/oauth/authorize",
+    GITHUB_API_BASE_URL="https://api.github.com/",
+    GITHUB_CLIENT_KWARGS={"scope": "user:email"},
+    FEDORA_CLIENT_KWARGS={"scope": "openid email profile"},
+    FEDORA_SERVER_METADATA_URL="https://id.fedoraproject.org/.well-known/openid-configuration",
+    GOOGLE_CLIENT_KWARGS={"scope": "openid email profile"},
+    GOOGLE_SERVER_METADATA_URL="https://accounts.google.com/.well-known/openid-configuration",
 )
 
 # Start with a basic logging configuration, which will be replaced by any user-

anitya/db/migrations/versions/feeaa70ead67_social_authentication_table.py

@@ -26,7 +26,6 @@
 
 import sqlalchemy as sa
 from alembic import op
-from social_sqlalchemy import storage
 from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.types import CHAR, TypeDecorator
 
@@ -140,7 +139,7 @@ def upgrade():
         "social_auth_partial",
         sa.Column("id", sa.Integer(), nullable=False),
         sa.Column("token", sa.String(length=32), nullable=True),
-        sa.Column("data", storage.JSONType(), nullable=True),
+        sa.Column("data", sa.JSON, nullable=True),
         sa.Column("next_step", sa.Integer(), nullable=True),
         sa.Column("backend", sa.String(length=32), nullable=True),
         sa.PrimaryKeyConstraint("id"),
@@ -155,7 +154,7 @@ def upgrade():
         "social_auth_usersocialauth",
         sa.Column("id", sa.Integer(), nullable=False),
         sa.Column("provider", sa.String(length=32), nullable=True),
-        sa.Column("extra_data", storage.JSONType(), nullable=True),
+        sa.Column("extra_data", sa.JSON, nullable=True),
         sa.Column("uid", sa.String(length=255), nullable=True),
         sa.Column("user_id", GUID(), nullable=False),
         sa.ForeignKeyConstraint(["user_id"], ["users.id"]),

anitya/db/models.py

@@ -977,8 +977,6 @@ class User(Base):
             able to log in.
         admin (bool): Determine if this user is an administrator. If True the user is
             administrator.
-        social_auth (sqlalchemy.orm.dynamic.AppenderQuery): The list of
-            :class:`social_flask_sqlalchemy.models.UserSocialAuth` entries for this user.
     """
 
     __tablename__ = "users"

anitya/sar.py

@@ -65,26 +65,6 @@ def main():
     users_list = []
     for user in users:
         user_dict = user.to_dict()
-        user_social_auths = db.Session.execute(
-            select(
-                text(
-                    "provider,extra_data,uid FROM social_auth_usersocialauth WHERE user_id = :val"
-                )
-            ),
-            {"val": str(user.id)},
-        )
-        user_dict["user_social_auths"] = []
-        # This part is working in postgresql, but in tests we are using sqlite
-        # which doesn't know the UUID type
-        # pylint: disable=not-an-iterable
-        for user_social_auth in user_social_auths:  # pragma: no cover
-            user_dict["user_social_auths"].append(
-                {
-                    "provider": user_social_auth["provider"],
-                    "extra_data": user_social_auth["extra_data"],
-                    "uid": user_social_auth["uid"],
-                }
-            )
         users_list.append(user_dict)
 
     json.dump(users_list, sys.stdout)

anitya/templates/login.html

@@ -3,21 +3,10 @@
 {% block title %}Login · Anitya{% endblock %}
 
 {% block body %}
-<div class="d-flex align-itmes-start gap-3">
-    {% for name, backend in available_backends.items() %}
-        {% if name == 'openid'%}
-            <form role="form" action="{{url_for('social.auth', backend=name)}}" method="post" class="d-flex">
-                <input name="openid_identifier" placeholder="https://id.openid.server" class="form-control me-2">
-                <button type="submit" class="btn btn-success">
-                    Login</button>
-            </form>
-        {% elif name == 'google-oauth2' %}
-            <a title="{{ Google }}" class="btn btn-primary" role="button"
-                href="{{url_for('social.auth', backend=name)}}">Google</a>
-        {% else %}
-            <a title="{{ name.capitalize() }}" class="btn btn-primary" role="button"
-                href="{{url_for('social.auth', backend=name)}}">{{ name.capitalize() }}</a>
-        {% endif %}
+<div class="socialaccount_ballot">
+  <ul class="socialaccount_providers list-inline">
+    {% for backend in available_backends %}
+    <li><a title="{{ backend }}" class="socialaccount_provider openid btn btn-default" href="{{ url_for('oauth.oauthlogin', name=backend.lower()) }}">{{ backend }}</a></li>
     {% endfor %}
 </div>
 {% endblock %}