[WD-26091] feat: Assign user|admin role based on launchpad team membership, upon SSO login (#230)

Author: muhammad-ali-pk

migrations/versions/14729444f282_.py

@@ -0,0 +1,25 @@
+"""Added role column to users table
+
+Revision ID: 14729444f282
+Revises: 1b6109065dba
+Create Date: 2025-08-26 17:15:11.023451
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "14729444f282"
+down_revision = "1b6109065dba"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column("users", sa.Column("role", sa.String(), default="user"))
+
+
+def downgrade():
+    op.drop_column("users", "role")

static/client/services/api/types/users.ts

@@ -5,6 +5,7 @@ export interface IUser {
   jobTitle: string;
   department: string;
   team: string;
+  role: string;
 }
 
 export interface IUsersResponse {

webapp/helper.py

@@ -27,6 +27,7 @@ def get_or_create_user_id(user):
             department=user.get("department"),
             job_title=user.get("jobTitle"),
             hrc_id=user.get("id"),
+            role=user.get("role"),
         )
 
     return user_exists.id

webapp/models.py

@@ -134,6 +134,7 @@ class User(db.Model, DateTimeMixin):
     department: str = Column(String)
     hrc_id: int = Column(Integer)
     job_title: str = Column(String)
+    role: str = Column(String)
 
     webpages = relationship("Webpage", back_populates="owner")
     reviewers = relationship("Reviewer", back_populates="user")

webapp/routes/user.py

@@ -112,6 +112,7 @@ def current_user():
                 "team": user.team,
                 "department": user.department,
                 "jobTitle": user.job_title,
+                "role": user.role,
             }
         ),
         200,

webapp/sso.py

@@ -6,16 +6,19 @@
 from flask_openid import OpenID
 
 from webapp.helper import get_or_create_user_id, get_user_from_directory_by_key
-from webapp.models import User
+from webapp.models import User, db
 
 SSO_LOGIN_URL = "https://login.ubuntu.com"
 # private teams like canonical are not returned in response atm
 # so temporarily need to add multiple subset public teams
+
+SSO_ADMIN_TEAM = "content-system-admins"
 SSO_TEAM = (
     "canonical",
     "canonical-content-people",
     "pga-admins",
     "canonical-webmonkeys",
+    SSO_ADMIN_TEAM,
 )
 DISABLE_SSO = os.environ.get("DISABLE_SSO") or os.environ.get(
     "FLASK_DISABLE_SSO"
@@ -45,20 +48,32 @@ def after_login(resp):
         if not (set(SSO_TEAM) & set(resp.extensions["lp"].is_member)):
             flask.abort(403)
 
+        # check if user is admin
+        role = (
+            "admin"
+            if SSO_ADMIN_TEAM in resp.extensions["lp"].is_member
+            else "user"
+        )
+
         # find the user in database
         user = User.query.filter_by(email=resp.email).first()
+        if user and user.role != role:
+            user.role = role
+            db.session.commit()
         if not user:
             # fetch user record from directory
             response = get_user_from_directory_by_key("email", resp.email)
 
             if response.status_code == 200:
                 user = response.json().get("data", {}).get("employees", [])[0]
+                user["role"] = role
                 # save user in users table
                 get_or_create_user_id(user)
 
         flask.session["openid"] = {
             "identity_url": resp.identity_url,
             "email": resp.email,
+            "role": role,
         }
 
         return flask.redirect(open_id.get_next_url())
@@ -101,3 +116,26 @@ def is_user_logged_in(*args, **kwargs):
         return flask.redirect("/login_page?next=" + flask.request.path)
 
     return is_user_logged_in
+
+
+def is_admin(func):
+    """
+    Decorator that checks if a user is an admin user
+    """
+
+    @functools.wraps(func)
+    def is_admin_user(*args, **kwargs):
+        if (
+            "openid" in flask.session
+            and flask.session.get("openid")["role"] == "admin"
+        ):
+            return func(*args, **kwargs)
+
+        return (
+            flask.jsonify(
+                {"error": "This operation requires admin privileges"}
+            ),
+            403,
+        )
+
+    return is_admin_user