Merge branch 'fix/TG-1119-mcp-error-scope-hardening' into 'enterprise'

fix(mcp): consistent error surfacing and scope validation

See merge request dkinternal/testgen/dataops-testgen!541

Author: ci bot

testgen/api/deps.py

@@ -6,7 +6,7 @@
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from sqlalchemy import select
 
-from testgen.common.auth import authorize_token, decode_jwt_token
+from testgen.common.auth import AuthError, authorize_token, decode_jwt_token
 from testgen.common.enums import PublicJobKey
 from testgen.common.models import Session, _current_session_wrapper, get_current_session
 from testgen.common.models.job_execution import JobExecution
@@ -48,7 +48,7 @@ def get_authorized_user(credentials: HTTPAuthorizationCredentials = _bearer_secu
 
     try:
         payload = decode_jwt_token(credentials.credentials)
-    except ValueError:
+    except AuthError:
         raise _invalid from None
 
     username = payload.get("username")
@@ -58,7 +58,7 @@ def get_authorized_user(credentials: HTTPAuthorizationCredentials = _bearer_secu
     session = get_current_session()
     try:
         return authorize_token(credentials.credentials, username, session)
-    except ValueError:
+    except AuthError:
         raise _invalid from None
 
 

testgen/common/auth.py

@@ -10,6 +10,10 @@
 LOG = logging.getLogger("testgen")
 
 
+class AuthError(Exception):
+    """Token authentication failed — invalid/expired token, unknown user, or revoked token."""
+
+
 def get_jwt_signing_key() -> bytes:
     """Decode the base64-encoded JWT signing key from settings."""
     return base64.b64decode(settings.JWT_HASHING_KEY_B64.encode("ascii"))
@@ -27,13 +31,13 @@ def create_jwt_token(username: str, expiry_seconds: int = 86400) -> str:
 def decode_jwt_token(token_str: str) -> dict:
     """Decode and validate a JWT token. Returns the payload dict.
 
-    Raises ValueError if the token is invalid or expired.
+    Raises ``AuthError`` if the token is invalid or expired.
     PyJWT auto-validates the standard ``exp`` claim during decode.
     """
     try:
         return jwt.decode(token_str, get_jwt_signing_key(), algorithms=["HS256"])
     except jwt.InvalidTokenError as e:
-        raise ValueError(f"Invalid token: {e}") from e
+        raise AuthError(f"Invalid token: {e}") from e
 
 
 def authorize_token(token_str: str, username: str, session):
@@ -48,11 +52,11 @@ def authorize_token(token_str: str, username: str, session):
 
     user = session.scalars(select(User).where(func.lower(User.username) == func.lower(username))).first()
     if user is None:
-        raise ValueError("User not found")
+        raise AuthError("User not found")
 
     token_record = session.scalars(select(OAuth2Token).where(OAuth2Token.access_token == token_str)).first()
     if token_record and token_record.access_token_revoked_at:
-        raise ValueError("Token has been revoked")
+        raise AuthError("Token has been revoked")
 
     return user
 

testgen/mcp/exceptions.py

@@ -20,6 +20,14 @@ class MCPUserError(Exception):
     """
 
 
+class MCPAuthenticationError(MCPUserError):
+    """Token authenticated at the transport but failed authorization at the tool boundary.
+
+    Raised when the token's user no longer exists or the token was revoked — distinct from
+    per-project permission denial. Carries a uniform, re-authenticate message for the client.
+    """
+
+
 class MCPPermissionDenied(MCPUserError):
     """Raised when access is denied due to insufficient project permissions."""
 

testgen/mcp/permissions.py

@@ -2,14 +2,17 @@
 
 import contextvars
 import functools
+import logging
 from collections.abc import Callable
 from dataclasses import dataclass
 
 from testgen.common.models.project_membership import ProjectMembership
 from testgen.common.models.user import User
-from testgen.mcp.exceptions import MCPPermissionDenied
+from testgen.mcp.exceptions import MCPAuthenticationError, MCPPermissionDenied
 from testgen.utils.plugins import PluginHook
 
+LOG = logging.getLogger("testgen")
+
 _NOT_SET = object()
 
 _mcp_username: contextvars.ContextVar[str | None] = contextvars.ContextVar("mcp_username", default=None)
@@ -83,7 +86,7 @@ def get_authorized_mcp_user() -> User:
     Checks user existence and token revocation status.
     Must be called within @with_database_session scope.
     """
-    from testgen.common.auth import authorize_token
+    from testgen.common.auth import AuthError, authorize_token
     from testgen.common.models import get_current_session
 
     username = _mcp_username.get()
@@ -92,7 +95,13 @@ def get_authorized_mcp_user() -> User:
 
     token_str = _mcp_token.get()
     session = get_current_session()
-    return authorize_token(token_str or "", username, session)
+    try:
+        return authorize_token(token_str or "", username, session)
+    except AuthError as err:
+        LOG.warning("MCP token authorization failed: %s", err)
+        raise MCPAuthenticationError(
+            "Authentication failed: your access token is no longer valid. Please sign in again."
+        ) from err
 
 
 def _compute_project_permissions(user: User, permission: str) -> ProjectPermissions:

testgen/mcp/server.py

@@ -9,7 +9,7 @@
 from starlette.applications import Starlette
 
 from testgen import settings
-from testgen.common.auth import decode_jwt_token
+from testgen.common.auth import AuthError, decode_jwt_token
 from testgen.mcp.permissions import set_mcp_token, set_mcp_username
 
 LOG = logging.getLogger("testgen")
@@ -65,7 +65,7 @@ async def verify_token(self, token: str) -> AccessToken | None:
                 scopes=[],
                 expires_at=int(payload["exp"]),
             )
-        except (ValueError, KeyError):
+        except (AuthError, KeyError):
             return None
 
 

testgen/mcp/tools/common.py

@@ -300,6 +300,22 @@ def parse_run_status_filter(value: str) -> list[JobStatus]:
     return statuses
 
 
+class FailureGroupBy(StrEnum):
+    """User-facing values accepted for the ``group_by`` argument on ``get_failure_summary``."""
+
+    TEST_TYPE = "test_type"
+    TABLE = "table"
+    COLUMN = "column"
+
+
+def parse_failure_group_by(value: str) -> FailureGroupBy:
+    try:
+        return FailureGroupBy(value)
+    except ValueError as err:
+        valid = ", ".join(g.value for g in FailureGroupBy)
+        raise MCPUserError(f"Invalid group_by `{value}`. Valid values: {valid}") from err
+
+
 def format_run_duration(started_at: datetime | None, completed_at: datetime | None) -> str | None:
     """Render an elapsed duration as ``Xs`` / ``Xm Ys`` / ``Xh Ym``. Returns ``None`` if either bound is missing."""
     if not started_at or not completed_at:
@@ -661,6 +677,52 @@ def resolve_notification(notification_id: str) -> NotificationSettings:
     return notif
 
 
+def resolve_aggregate_scope(
+    project_code: str | None,
+    test_suite_id: str | None = None,
+    table_group_id: str | None = None,
+) -> list[str]:
+    """Validate optional project / test-suite / table-group scope for a cross-run aggregation.
+
+    Resolves any supplied suite or table group (existence + access via the
+    ``resolve_*`` helpers), and — when ``project_code`` is also given — requires the
+    resolved entity to belong to it, raising a clear error on a cross-project mismatch
+    so the query never silently returns empty. Returns the project codes to scope the
+    aggregation to.
+    """
+    perms = get_project_permissions()
+    if project_code:
+        perms.verify_access(project_code, not_found=MCPResourceNotAccessible("Project", project_code))
+
+    scoped_projects: set[str] = set()
+    if test_suite_id:
+        suite = resolve_test_suite(test_suite_id)
+        if project_code and suite.project_code != project_code:
+            raise MCPUserError(
+                f"Test suite `{test_suite_id}` belongs to project `{suite.project_code}`, not `{project_code}`."
+            )
+        scoped_projects.add(suite.project_code)
+    if table_group_id:
+        table_group = resolve_table_group(table_group_id)
+        if project_code and table_group.project_code != project_code:
+            raise MCPUserError(
+                f"Table group `{table_group_id}` belongs to project `{table_group.project_code}`, not `{project_code}`."
+            )
+        scoped_projects.add(table_group.project_code)
+
+    if len(scoped_projects) > 1:
+        # Suite and table group resolve to different projects (only reachable when no
+        # project_code pins the scope). The two filters would AND to an empty result —
+        # reject rather than silently return nothing.
+        raise MCPUserError("The test suite and table group belong to different projects — narrow to one scope.")
+
+    if project_code:
+        return [project_code]
+    if scoped_projects:
+        return list(scoped_projects)
+    return perms.allowed_codes
+
+
 # Notification event-type labels.
 
 class NotificationEventLabel(StrEnum):

testgen/mcp/tools/discovery.py

@@ -4,6 +4,7 @@
 from testgen.common.models.project import Project
 from testgen.common.models.test_run import TestRun
 from testgen.common.models.test_suite import TestSuite
+from testgen.mcp.exceptions import MCPResourceNotAccessible
 from testgen.mcp.permissions import get_project_permissions, mcp_permission
 from testgen.mcp.tools.common import DocGroup, resolve_table_group, validate_limit, validate_page
 from testgen.mcp.tools.markdown import MdDoc
@@ -63,7 +64,7 @@ def list_test_suites(project_code: str) -> str:
         return "Missing required parameter `project_code`."
 
     perms = get_project_permissions()
-    perms.verify_access(project_code, not_found=f"No test suites found for project `{project_code}`.")
+    perms.verify_access(project_code, not_found=MCPResourceNotAccessible("Project", project_code))
 
     summaries = TestSuite.select_summary(project_code)
 

testgen/mcp/tools/test_results.py

@@ -12,11 +12,14 @@
 from testgen.mcp.permissions import get_project_permissions, mcp_permission
 from testgen.mcp.tools.common import (
     DocGroup,
+    FailureGroupBy,
     format_page_footer,
     format_page_info,
+    parse_failure_group_by,
     parse_result_status,
     parse_since_arg,
     parse_uuid,
+    resolve_aggregate_scope,
     resolve_test_type,
     validate_limit,
     validate_page,
@@ -27,6 +30,12 @@
 
 _DEFAULT_SEARCH_STATUSES = [TestResultStatus.Failed, TestResultStatus.Warning]
 
+_MODEL_GROUP_COLUMN = {
+    FailureGroupBy.TEST_TYPE: "test_type",
+    FailureGroupBy.TABLE: "table_name",
+    FailureGroupBy.COLUMN: "column_names",
+}
+
 
 @with_database_session
 @mcp_permission("view")
@@ -167,20 +176,20 @@ def get_failure_summary(
         group_by: Group failures by 'test_type', 'table', or 'column' (default: 'test_type').
     """
     perms = get_project_permissions()
+    group = parse_failure_group_by(group_by)
 
     if not any((job_execution_id, test_suite_id, since)):
         raise MCPUserError(
             "Provide 'job_execution_id' for a single run, or 'test_suite_id' or 'project_code' "
             "to aggregate across runs. 'since' is required when 'test_suite_id' is not provided."
         )
-    if group_by in ("table", "column") and not (job_execution_id or test_suite_id):
+    if group in (FailureGroupBy.TABLE, FailureGroupBy.COLUMN) and not (job_execution_id or test_suite_id):
         raise MCPUserError(
-            f"'{group_by}' grouping requires a single-suite scope. "
+            f"'{group}' grouping requires a single-suite scope. "
             "Provide 'job_execution_id' or 'test_suite_id'."
         )
 
-    model_group_map = {"table": "table_name", "column": "column_names"}
-    model_group_by = model_group_map.get(group_by, group_by)
+    model_group_by = _MODEL_GROUP_COLUMN[group]
 
     scope_label: str
     test_run_id = None
@@ -197,15 +206,7 @@ def get_failure_summary(
         scope_label = f"run `{job_execution_id}`"
         project_codes = perms.allowed_codes
     else:
-        if project_code:
-            perms.verify_access(project_code, not_found=MCPResourceNotAccessible("Project", project_code))
-            project_codes = [project_code]
-        else:
-            project_codes = perms.allowed_codes
-        if test_suite_uuid is not None:
-            suite = TestSuite.get_regular(test_suite_uuid)
-            if suite is None or not perms.has_access(suite.project_code):
-                raise MCPResourceNotAccessible("Test suite", test_suite_id)
+        project_codes = resolve_aggregate_scope(project_code, test_suite_id=test_suite_id)
         scope_parts = []
         if project_code:
             scope_parts.append(f"project `{project_code}`")
@@ -227,22 +228,22 @@ def get_failure_summary(
         return f"No confirmed failures found for {scope_label}."
 
     total = sum(row[-1] for row in failures)
-    if group_by == "test_type":
+    if group is FailureGroupBy.TEST_TYPE:
         type_names = {tt.test_type: tt.test_name_short for tt in TestType.select_where(TestType.active == "Y")}
 
     doc = MdDoc()
     doc.heading(1, f"Failure Summary — {scope_label}")
     doc.text(f"**Total confirmed failures (Failed + Warning):** {total}")
 
-    if group_by == "test_type":
+    if group is FailureGroupBy.TEST_TYPE:
         headers = ["Test Type", "Severity", "Count"]
         rows = []
         for row in failures:
             code, status, count = row[0], row[1], row[-1]
             name = type_names.get(code, code)
             severity = status.value if status else "Unknown"
             rows.append([name, severity, count])
-    elif group_by == "column":
+    elif group is FailureGroupBy.COLUMN:
         headers = ["Column", "Count"]
         rows = []
         for row in failures:
@@ -253,9 +254,9 @@ def get_failure_summary(
         headers = ["Table Name", "Count"]
         rows = [[row[0], row[-1]] for row in failures]
 
-    doc.table(headers, rows, code=[0] if group_by == "table" else None)
+    doc.table(headers, rows, code=[0] if group is FailureGroupBy.TABLE else None)
 
-    if group_by == "test_type":
+    if group is FailureGroupBy.TEST_TYPE:
         doc.text(
             "Check `testgen://test-types` to understand what each test type checks "
             "and `get_test_type(test_type='...')` to fetch more details."
@@ -450,12 +451,9 @@ def get_failure_trend(
         valid = ", ".join(v.value for v in BucketInterval)
         raise MCPUserError(f"Invalid `bucket`: `{bucket}`. Valid values: {valid}") from err
 
-    perms = get_project_permissions()
-    if project_code:
-        perms.verify_access(project_code, not_found=MCPResourceNotAccessible("Project", project_code))
-        project_codes = [project_code]
-    else:
-        project_codes = perms.allowed_codes
+    project_codes = resolve_aggregate_scope(
+        project_code, test_suite_id=test_suite_id, table_group_id=table_group_id
+    )
 
     anchor_today = datetime.now(UTC).date()
     if exclude_today: