feat: migrate sso auth to c-idp

Author: muhammad-ali-pk

.env

@@ -30,3 +30,7 @@ FLASK_TRINO_SF_PRIVATE_KEY=trino_private_key
 # FLASK_DISABLE_AUTH_FOR_TESTS=1
 
 
+# C-IDP Creds
+FLASK_SSO_CLIENT_ID=client-id
+FLASK_SSO_CLIENT_SECRET=client-secret
+FLASK_SSO_PROVIDER=oidc-provider-url

charm/charmcraft.yaml

@@ -38,4 +38,7 @@ config:
       default: false
     trino-sf:
       type: secret
-      description: Trino credentials, must contain (project-id, private-key-id, client-email, client-id, private-key)
+      description: Trino credentials, must contain (project-id, private-key-id, client-email, client-id, private-key)
+    sso:
+      type: secret
+      description: SSO credentials, must contain (client-id, client-secret, provider)

requirements.txt

@@ -22,4 +22,5 @@ google-auth==2.40.3
 cryptography==46.0.1
 setuptools==79.0.1
 black==25.1.0
-flake8==7.1.1
+flake8==7.1.1
+Authlib==1.6.6

webapp/sso.py

@@ -2,43 +2,74 @@
 import os
 
 import flask
-from django_openid_auth.teams import TeamsRequest, TeamsResponse
-from flask_openid import OpenID
+from authlib.integrations.flask_client import OAuth
+import requests
 
+from webapp.views import get_users_data
 
-SSO_LOGIN_URL = "https://login.ubuntu.com"
 SSO_TEAM = "canonical-content-people"
+LAUNCHPAD_API_URL = "https://api.launchpad.net/1.0"
 
 
 def init_sso(app):
-    open_id = OpenID(
-        store_factory=lambda: None,
-        safe_roots=[],
-        extension_responses=[TeamsResponse],
+
+    oauth = OAuth(app)
+
+    oauth.register(
+        "canonical",
+        client_id=os.getenv("SSO_CLIENT_ID"),
+        client_secret=os.getenv("SSO_CLIENT_SECRET"),
+        server_metadata_url=os.getenv("SSO_PROVIDER"),
+        client_kwargs={
+            "token_endpoint_auth_method": "client_secret_post",
+            "scope": "openid profile email",
+        },
     )
 
-    @app.route("/login", methods=["GET", "POST"])
-    @open_id.loginhandler
+    @app.route("/login")
     def login():
         if "openid" in flask.session:
-            return flask.redirect(open_id.get_next_url())
+            return flask.redirect(flask.request.args.get("next") or "/manager")
+
+        redirect_uri = flask.url_for("oauth_callback", _external=True)
+        return oauth.canonical.authorize_redirect(redirect_uri)
 
-        teams_request = TeamsRequest(query_membership=[SSO_TEAM])
-        return open_id.try_login(
-            SSO_LOGIN_URL, ask_for=["email"], extensions=[teams_request]
+    @app.route("/auth/callback")
+    def oauth_callback():
+        token = oauth.canonical.authorize_access_token()
+        users, status_code = get_users_data(token["userinfo"]["name"])
+
+        if status_code != 200 or not users:
+            flask.abort(
+                403, description="Failed to fetch user data from directory."
+            )
+
+        response = requests.get(
+            f"{LAUNCHPAD_API_URL}/~{users[0]['launchpadId']}/super_teams",
         )
 
-    @open_id.after_login
-    def after_login(resp):
-        if SSO_TEAM not in resp.extensions["lp"].is_member:
-            flask.abort(403)
+        if response.status_code != 200:
+            flask.abort(
+                403, description="Failed to fetch Launchpad team memberships."
+            )
+
+        memberships = response.json().get("entries", [])
+        if SSO_TEAM not in [team["name"] for team in memberships]:
+            flask.abort(
+                403,
+                description=(
+                    "Please make sure you are a member of the "
+                    "canonical-content-people team on Launchpad."
+                ),
+            )
 
         flask.session["openid"] = {
-            "identity_url": resp.identity_url,
-            "email": resp.email,
+            "identity_url": token["userinfo"]["iss"],
+            "email": token["userinfo"]["email"],
+            "fullname": token["userinfo"]["name"],
         }
 
-        return flask.redirect(open_id.get_next_url())
+        return flask.redirect(flask.request.args.get("next") or "/manager")
 
 
 def login_required(func):

webapp/views.py

@@ -474,20 +474,21 @@ def delete_redirect(redirect_path):
     return jsonify({}), 204
 
 
-def get_users(username: str):
+def get_users_data(username: str):
     query = """
-    query($name: String!) {
-        employees(filter: { contains: { name: $name }}) {
-            id
-            firstName
-            surname
-            email
-            team
-            department
-            jobTitle
+        query($name: String!) {
+            employees(filter: { contains: { name: $name }}) {
+                id
+                firstName
+                surname
+                email
+                team
+                department
+                jobTitle
+                launchpadId
+            }
         }
-    }
-    """
+        """
 
     headers = {
         "Authorization": "token "
@@ -506,6 +507,14 @@ def get_users(username: str):
 
     if response.status_code == 200:
         users = response.json().get("data", {}).get("employees", [])
+        return users, 200
+
+    return None, response.status_code
+
+
+def get_users(username: str):
+    users, status_code = get_users_data(username)
+    if status_code == 200:
         return jsonify(list(users))
     return jsonify({"error": "Failed to fetch users"}), 500