Team-Level API Token Scoping with Public-Only Token Support (#1177)

* team level resources scoping for api tokens

Signed-off-by: Keval Mahajan <[email protected]>

* public-only token support

Signed-off-by: Keval Mahajan <[email protected]>

* regex pattern matching covering all required apis

Signed-off-by: Keval Mahajan <[email protected]>

* linting issues

Signed-off-by: Keval Mahajan <[email protected]>

* minor changes

Signed-off-by: Keval Mahajan <[email protected]>

* web linting

Signed-off-by: Keval Mahajan <[email protected]>

---------

Signed-off-by: Keval Mahajan <[email protected]>

Author: kevalmahajan

mcpgateway/middleware/token_scoping.py

@@ -59,6 +59,11 @@ async def create_token(
     """
     service = TokenCatalogService(db)
 
+    # if not request.team_id:
+    #     raise HTTPException(
+    #         status_code=status.HTTP_400_BAD_REQUEST, detail="team_id is required. Please select a specific team before creating a token. You cannot create tokens while viewing 'All Teams'."
+    #     )
+
     # Convert request to TokenScope if provided
     scope = None
     if request.scope:
@@ -78,7 +83,7 @@ async def create_token(
             scope=scope,
             expires_in_days=request.expires_in_days,
             tags=request.tags,
-            team_id=getattr(request, "team_id", None),
+            team_id=request.team_id,
         )
 
         # Create TokenResponse for the token info

mcpgateway/routers/tokens.py

@@ -299,99 +299,124 @@ async def create_token(
         tags: Optional[List[str]] = None,
         team_id: Optional[str] = None,
     ) -> tuple[EmailApiToken, str]:
-        """Create a new API token for user or team.
+        """
+        Create a new API token with team-level scoping and additional configurations.
+
+        This method generates a JWT-based API token with team-level scoping and optional security configurations,
+        such as expiration, permissions, IP restrictions, and usage limits. The token is associated with a user
+        and a specific team, ensuring access control and multi-tenancy support.
+
+        The function will:
+        - Validate the existence of the user.
+        - Ensure the user is an active member of the specified team.
+        - Verify that the token name is unique for the user+team combination.
+        - Generate a JWT with the specified scoping parameters (e.g., permissions, IP, etc.).
+        - Store the token in the database with the relevant details and return the token and raw JWT string.
 
         Args:
-            user_email: Owner's email address
-            name: Human-readable token name
-            description: Optional token description
-            scope: Token scoping configuration
-            expires_in_days: Optional expiry in days
-            tags: Optional organizational tags
-            team_id: Optional team ID for team-scoped tokens
+            user_email (str): The email address of the user requesting the token.
+            name (str): A unique, human-readable name for the token (must be unique per user+team).
+            description (Optional[str]): A description for the token (default is None).
+            scope (Optional[TokenScope]): The scoping configuration for the token, including permissions,
+                server ID, IP restrictions, etc. (default is None).
+            expires_in_days (Optional[int]): The expiration time in days for the token (None means no expiration).
+            tags (Optional[List[str]]): A list of organizational tags for the token (default is an empty list).
+            team_id (Optional[str]): The team ID to which the token should be scoped. This is required for team-level scoping.
 
         Returns:
-            Tuple of (EmailApiToken, raw_token_string)
+            tuple[EmailApiToken, str]: A tuple where the first element is the `EmailApiToken` database record and
+            the second element is the raw JWT token string. The `EmailApiToken` contains the database record with the
+            token details.
 
         Raises:
-            ValueError: If user not found, token name exists, or team access denied
+            ValueError: If any of the following validation checks fail:
+                - The `user_email` does not correspond to an existing user.
+                - The `team_id` is missing or the user is not an active member of the specified team.
+                - A token with the same name already exists for the given user and team.
+                - Invalid token configuration (e.g., invalid expiration date).
 
         Examples:
             >>> # This method requires database operations, shown for reference
             >>> service = TokenCatalogService(None)  # Would use real DB session
             >>> # token, raw_token = await service.create_token(...)
             >>> # Returns (EmailApiToken, raw_token_string) tuple
         """
+        # # Enforce team-level scoping requirement
+        # if not team_id:
+        #     raise ValueError("team_id is required for token creation. " "Please select a specific team before creating a token. " "You cannot create tokens while viewing 'All Teams'.")
+
         # Validate user exists
         user = self.db.execute(select(EmailUser).where(EmailUser.email == user_email)).scalar_one_or_none()
 
         if not user:
             raise ValueError(f"User not found: {user_email}")
 
-        # Validate team access if team_id is provided
+        # Validate team exists and user is active member
         if team_id:
             # First-Party
             from mcpgateway.db import EmailTeam, EmailTeamMember  # pylint: disable=import-outside-toplevel
 
             # Check if team exists
             team = self.db.execute(select(EmailTeam).where(EmailTeam.id == team_id)).scalar_one_or_none()
+
             if not team:
                 raise ValueError(f"Team not found: {team_id}")
 
-            # Check if user is a team OWNER
+            # Verify user is an active member of the team
             membership = self.db.execute(
-                select(EmailTeamMember).where(and_(EmailTeamMember.team_id == team_id, EmailTeamMember.user_email == user_email, EmailTeamMember.role == "owner", EmailTeamMember.is_active.is_(True)))
+                select(EmailTeamMember).where(and_(EmailTeamMember.team_id == team_id, EmailTeamMember.user_email == user_email, EmailTeamMember.is_active.is_(True)))
             ).scalar_one_or_none()
 
             if not membership:
-                raise ValueError(f"Only team owners can create API keys for team {team_id}")
-
-        # Check for duplicate token name (scoped by user and team)
-        name_check_conditions = [EmailApiToken.user_email == user_email, EmailApiToken.name == name, EmailApiToken.is_active.is_(True)]
+                raise ValueError(f"User {user_email} is not an active member of team {team_id}. " f"Only team members can create tokens for the team.")
 
-        if team_id:
-            name_check_conditions.append(EmailApiToken.team_id == team_id)
-        else:
-            name_check_conditions.append(EmailApiToken.team_id.is_(None))
-
-        existing_token = self.db.execute(select(EmailApiToken).where(and_(*name_check_conditions))).scalar_one_or_none()
+        # Check for duplicate active token name for this user+team
+        existing_token = self.db.execute(
+            select(EmailApiToken).where(and_(EmailApiToken.user_email == user_email, EmailApiToken.name == name, EmailApiToken.team_id == team_id, EmailApiToken.is_active.is_(True)))
+        ).scalar_one_or_none()
 
         if existing_token:
-            scope_desc = f" for team {team_id}" if team_id else ""
-            raise ValueError(f"Token name '{name}' already exists{scope_desc}")
+            raise ValueError(f"Token with name '{name}' already exists for user {user_email} in team {team_id}. Please choose a different name.")
 
-        # Calculate expiry
+        # CALCULATE EXPIRATION DATE
         expires_at = None
         if expires_in_days:
             expires_at = utc_now() + timedelta(days=expires_in_days)
 
-        # Generate JWT token with proper claims and user admin status
-        raw_token = await self._generate_token(user_email=user_email, team_id=team_id, expires_at=expires_at, scope=scope, user=user)
+        # Generate JWT token with all necessary claims
+        raw_token = await self._generate_token(user_email=user_email, team_id=team_id, expires_at=expires_at, scope=scope, user=user)  # Pass user object to include admin status
+
+        # Hash token for secure storage
         token_hash = self._hash_token(raw_token)
 
-        # Create token record
+        # Create database record
         api_token = EmailApiToken(
+            id=str(uuid.uuid4()),
             user_email=user_email,
+            team_id=team_id,  # Store team association
             name=name,
-            token_hash=token_hash,
             description=description,
+            token_hash=token_hash,  # Store hash, not raw token
             expires_at=expires_at,
             tags=tags or [],
-            team_id=team_id,
+            # Store scoping information
             server_id=scope.server_id if scope else None,
             resource_scopes=scope.permissions if scope else [],
             ip_restrictions=scope.ip_restrictions if scope else [],
             time_restrictions=scope.time_restrictions if scope else {},
             usage_limits=scope.usage_limits if scope else {},
+            # Token status
+            is_active=True,
+            created_at=utc_now(),
+            last_used=None,
         )
 
         self.db.add(api_token)
         self.db.commit()
         self.db.refresh(api_token)
 
-        scope_desc = f" for team {team_id}" if team_id else ""
-        logger.info(f"Created API token '{name}' for user {user_email}{scope_desc}")
-
+        token_type = f"team-scoped (team: {team_id})" if team_id else "public-only"
+        logger.info(f"Created {token_type} API token '{name}' for user {user_email}. " f"Token ID: {api_token.id}, Expires: {expires_at or 'Never'}")
         return api_token, raw_token
 
     async def list_user_tokens(self, user_email: str, include_inactive: bool = False, limit: int = 100, offset: int = 0) -> List[EmailApiToken]: