From ca4447b7f89a5f51275da3124e383d2af1405b6c Mon Sep 17 00:00:00 2001 From: Alexliu Date: Fri, 4 Sep 2026 01:25:29 +0800 Subject: [PATCH 01/10] feat(connectors): let team members edit a team-shared Custom API GET and PUT /api/custom-apis/{api_id} resolve a caller with no personal association row through the connector access hook instead of 404ing outright, and the edit right is granted either by the caller's own link row or by a team verdict that grants edit. A caller admitted by a verdict carries the team-owned stand-in in place of an association row. That stand-in holds no persistent state, so a payload carrying is_active from such a caller is refused with 400 rather than writing a shadow attribute the response would then read back. The verdict is resolved before the definition row's lock is taken, and the lock waits, so the payloads that take it re-establish both halves of the gate's decision afterwards: the caller's link row is re-read and the verdict re-resolved, and a request whose authorization no longer holds is rolled back and refused. A payload that writes only the caller's own link row takes no lock and needs no re-check. A raising hook surfaces as the status the seam declares rather than as a generic 500, on every call site this route makes. --- src/xagent/web/api/custom_api.py | 338 ++++++++++++++++++++++++------- 1 file changed, 264 insertions(+), 74 deletions(-) diff --git a/src/xagent/web/api/custom_api.py b/src/xagent/web/api/custom_api.py index 461a5052a..97440e339 100644 --- a/src/xagent/web/api/custom_api.py +++ b/src/xagent/web/api/custom_api.py @@ -7,13 +7,14 @@ import logging from datetime import datetime -from typing import Any, Dict, List, Optional, cast +from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, cast from fastapi import APIRouter, Depends, HTTPException, status from pydantic import BaseModel, Field from sqlalchemy.orm import Session from ...core.tools.adapters.vibe.connector_runtime import ( + ConnectorRuntimeError, validate_runtime_config_declaration, ) from ...core.utils.encryption import encrypt_value @@ -22,6 +23,10 @@ from ..models.database import get_db from ..models.user import User +if TYPE_CHECKING: + from ..services.connector_team_scope import ConnectorAccess + from .mcp import _TeamOwnedUserApi + logger = logging.getLogger(__name__) @@ -111,7 +116,7 @@ class Config: def _db_api_to_response( api: CustomApi, - user_api: UserCustomApi, + user_api: "UserCustomApi | _TeamOwnedUserApi", ) -> CustomApiResponse: """Convert database CustomApi to response model with masked env values.""" @@ -259,30 +264,122 @@ async def create_custom_api( return _db_api_to_response(new_api, user_api) -@custom_api_router.get("/{api_id}", response_model=CustomApiResponse) -async def get_custom_api( +def _http_from_connector_runtime(exc: ConnectorRuntimeError) -> HTTPException: + """Map the connector team seam's typed error onto this module's HTTP answer. + + Four call sites need it -- ``get_custom_api``, and ``update_custom_api`` + for its pre-lock resolution, its post-lock re-check and its rename hook -- + so that every one of them answers with the status and message the seam + declares instead of letting the error reach the generic handler as a 500. + Each call site wraps only the seam call it makes, rather than the whole + route body, so the mapping itself lives here in one place. + """ + return HTTPException(status_code=exc.status_code, detail=exc.safe_message) + + +def _resolve_custom_api_for_request( + db: Session, + user_id: int, api_id: int, - current_user: User = Depends(get_current_user), - db: Session = Depends(get_db), -) -> CustomApiResponse: - """Get a specific Custom API by ID.""" + *, + skip_resolution_when: "Callable[[UserCustomApi], bool] | None" = None, +) -> "tuple[UserCustomApi | _TeamOwnedUserApi, CustomApi, ConnectorAccess | None]": + """Resolve the caller's association, the definition row, and the caller's + team access verdict, for ``GET``/``PUT /api/custom-apis/{api_id}``. + + Looks up the caller's own personal link row first, with the same query + both routes have always run. When that row exists and its ``custom_api`` + relationship resolves, the association and the definition row both come + from it and nothing else runs. When it does not -- no row, or a row whose + relationship is unexpectedly empty -- the definition row is looked up on + its own, because a team-owned API's shared row must still be found even + though this caller has no personal link to it, and the caller's team + access verdict decides what happens next: + + - no working personal row and no team access (``access is None``) -> 404, + the same outcome every caller without an association has always gotten. + - no working personal row but the caller's team links the API -> the + ``_TeamOwnedUserApi`` stand-in takes the association's place. It carries + the caller's own id and the flag defaults that apply when no personal + row exists, so the response contract stays the shape it always was. + + ``skip_resolution_when`` lets a caller declare when its own working + personal row already decides the answer on its own, so resolving a verdict + would only add an unnecessary hook call: ``get_custom_api`` passes a + predicate that is always true, because it never reads the verdict at all + and a personal row -- edit rights or not -- already decides what it + returns; ``update_custom_api`` passes one that checks ``can_edit``, + because only a personal row with ``can_edit=True`` decides the edit answer + on its own -- a ``can_edit=False`` row does not, since a granting team + verdict can still widen it. Left unset (the default), resolution is never + skipped, which is what a caller with no working personal row always needs + -- the verdict is the gate there and must stay fail-closed. + + Raises ``ConnectorRuntimeError`` when access resolution itself fails; + callers translate that into an ``HTTPException``. + """ + from ..services.connector_team_scope import resolve_one_connector_access_or_raise + from .mcp import _TeamOwnedUserApi user_api = ( db.query(UserCustomApi) .filter( UserCustomApi.custom_api_id == api_id, - UserCustomApi.user_id == current_user.id, + UserCustomApi.user_id == user_id, ) .first() ) + if user_api is not None and user_api.custom_api is not None: + api: Optional[CustomApi] = user_api.custom_api + else: + user_api = None + api = db.query(CustomApi).filter(CustomApi.id == api_id).first() - if not user_api or not user_api.custom_api: + already_decided = user_api is not None and ( + skip_resolution_when is not None and skip_resolution_when(user_api) + ) + + access: "ConnectorAccess | None" = None + if api is not None and not already_decided: + access = resolve_one_connector_access_or_raise( + db, int(user_id), ("custom_api", int(api.id)) + ) + + if user_api is None and access is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Custom API not found", ) - return _db_api_to_response(user_api.custom_api, user_api) + resolved_user_api: "UserCustomApi | _TeamOwnedUserApi" = ( + user_api if user_api is not None else _TeamOwnedUserApi(int(user_id)) + ) + return resolved_user_api, cast(CustomApi, api), access + + +@custom_api_router.get("/{api_id}", response_model=CustomApiResponse) +def get_custom_api( + api_id: int, + current_user: User = Depends(get_current_user), + db: Session = Depends(get_db), +) -> CustomApiResponse: + """Get a specific Custom API by ID.""" + + try: + # This route never reads the team verdict at all (see + # _db_api_to_response), so a working personal row -- edit rights or + # not -- already decides everything this route returns; resolving a + # verdict for such a caller would only add an unnecessary hook call. + user_api, api, _team_access = _resolve_custom_api_for_request( + db, + int(current_user.id), + api_id, + skip_resolution_when=lambda _user_api: True, + ) + except ConnectorRuntimeError as exc: + raise _http_from_connector_runtime(exc) from exc + + return _db_api_to_response(api, user_api) @custom_api_router.put("/{api_id}", response_model=CustomApiResponse) @@ -294,25 +391,48 @@ def update_custom_api( ) -> CustomApiResponse: """Update an existing Custom API.""" - user_api = ( - db.query(UserCustomApi) - .filter( - UserCustomApi.custom_api_id == api_id, - UserCustomApi.user_id == current_user.id, + try: + # A personal row with can_edit=True already decides the edit answer on + # its own (the gate just below), so resolving a verdict for that row + # would only add an unnecessary hook call. A can_edit=False personal + # row does not decide it, because a granting team verdict can still + # widen it, so that caller's verdict is resolved. + # + # The definition row this resolution loads is deliberately dropped: + # the fresh single-table read further down is the one every field + # below reads and writes. + user_api, _pre_lock_api, team_access = _resolve_custom_api_for_request( + db, + int(current_user.id), + api_id, + skip_resolution_when=lambda ua: bool(ua.can_edit), ) - .first() - ) - - if not user_api or not user_api.custom_api: + except ConnectorRuntimeError as exc: + raise _http_from_connector_runtime(exc) from exc + + # Two independent grants of the edit right: the caller's own link row, or + # a team verdict that grants edit on a connector the caller's team links. + # A caller with no personal row at all reaches this with the stand-in, + # whose can_edit is False, so only the verdict can admit that caller. + is_stand_in = not isinstance(user_api, UserCustomApi) + if not ( + bool(user_api.can_edit) or (team_access is not None and team_access.can_edit) + ): raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail="Custom API not found", + status_code=status.HTTP_403_FORBIDDEN, + detail="You do not have permission to edit this Custom API", ) - if not user_api.can_edit: + # is_active lives on the personal association row; a caller with no + # personal row (the stand-in) has none to hold it, so a payload carrying + # it must be rejected outright -- writing it onto the stand-in would only + # set a shadowing instance attribute that persists nothing, and the + # response below would then read that shadow back and report a change that + # never happened. + if is_stand_in and api_data.is_active is not None: raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail="You do not have permission to edit this Custom API", + status_code=status.HTTP_400_BAD_REQUEST, + detail="No personal connection exists to configure is_active for this API", ) # Which row a request writes decides which row it locks. Every field of @@ -333,22 +453,25 @@ def update_custom_api( # cases that did not need it and skips it in none that do. fields_set = api_data.model_fields_set writes_definition_row = bool(fields_set - {"is_active"}) - # This flag also gates the post-lock re-read of the caller's link row - # further down, not only the lock itself: adding a future field to the - # ``{"is_active"}`` exclusion set above -- because it too writes only - # the link row -- would silently skip that re-authorization as well, - # not just the lock, for any payload that sets only that field. + # This flag also gates the whole post-lock re-authorization further down + # -- the fresh read of the caller's link row and the fresh resolution of + # the team verdict -- not only the lock itself: adding a future field to + # the ``{"is_active"}`` exclusion set above, because it too writes only + # the link row, would silently skip that re-authorization as well, not + # just the lock, for any payload that sets only that field. # A fresh single-table read of the definition row, on both paths. The - # read above comes through the personal link row's relationship and - # cannot itself address just this table; this is a separate statement, - # so a row deleted between the two yields None here (handled as the - # same 404) rather than surfacing as an unrelated error out of the - # write path or out of ``db.refresh`` below. ``populate_existing()`` - # makes this statement's row the one the rest of this route reads and - # responds with: without it the already-identity-mapped instance the - # relationship loaded would be returned unrefreshed, and every field - # below would still be that earlier snapshot. + # gate above reached this row through the personal link row's + # relationship, or through a plain lookup for a caller who has no + # personal link row, and neither is a statement this route can add a + # locking clause to; this is a separate statement, so a row deleted + # between the two yields None here (handled as the same 404) rather + # than surfacing as an unrelated error out of the write path or out of + # ``db.refresh`` below. ``populate_existing()`` makes this statement's + # row the one the rest of this route reads and responds with: without + # it the already-identity-mapped instance the gate loaded would be + # returned unrefreshed, and every field below would still be that + # earlier snapshot. # # ``FOR UPDATE`` is added only on the path that writes this row, so a # request that writes it still waits for another request holding it. @@ -386,35 +509,42 @@ def update_custom_api( if writes_definition_row: # The access gate above ran before the lock statement; the lock - # statement waits. Everything this route decided from the gate's - # ``UserCustomApi`` row -- that the caller still has a link to this - # connector at all, and that the link grants edit -- was therefore + # statement waits. Everything the gate decided -- that this caller + # may edit this connector, whether that came from the caller's own + # link row or from the team access verdict -- was therefore # established before a wait of unbounded length, and nothing has # re-established it since. A supported admin user deletion # (``admin_users.py``, which removes a user's association rows and - # leaves every definition row standing) or a connector-team delete - # that removes only the caller's link can commit inside that wait. - # The request would then resume on a row that is gone, write the - # shared definition row anyway, commit it, and only fail afterwards - # in response construction -- an HTTP 500 over a durable shared - # mutation the caller was no longer authorized to make. + # leaves every definition row standing), a connector-team delete + # that removes only the caller's link, or a revocation of the + # caller's team access inside the installing application's own + # tables can each commit inside that wait. The request would then + # resume unauthorized, write the shared definition row anyway, and + # commit it -- a durable shared mutation the caller was no longer + # allowed to make. + # + # So both halves of the gate's decision are re-established here from + # fresh reads and the same combination is recomputed: the caller's + # own link row is re-read, and the team verdict is re-resolved when + # it is the half that has to grant the edit. A payload that writes + # only the caller's own link row skips this block entirely, and + # correctly so -- it took no lock above, so its gate was never + # separated from its writes by a wait of unbounded length. # # ``populate_existing()`` on the definition query above refreshes # the row of that statement and nothing else, so it does not cover - # this: the link row needs its own statement. This one is a + # the link row: that needs its own statement. This one is a # single-table read of the caller's link, with # ``populate_existing()`` so its columns are overwritten with the - # database's current values rather than the ones the gate loaded, - # and the object it returns replaces ``user_api`` for the rest of - # the route -- the link-row write below and the response both read it. - # A revocation that commits after this statement is the window + # database's current values rather than the ones the gate loaded. + # When it returns a row, that row replaces ``user_api`` for the rest + # of the route -- the link-row write below and the response both read + # it; when it returns nothing, the caller had no personal row to begin + # with and keeps the stand-in it was resolved with. + # A revocation that commits after these statements is the window # this route had before it took any lock at all: in-process, with - # no wait in it. Closing that one needs the authorization fence - # designed for the team-edit changes and is not attempted here. - # - # Same order as the gate, so the same request gets the same answer - # it would have got had it arrived a moment later: gone is a 404, - # present but no longer permitted is a 403. + # no wait in it. Closing that one needs an authorization fence of + # its own and is not attempted here. current_user_api = ( db.query(UserCustomApi) .filter( @@ -424,22 +554,72 @@ def update_custom_api( .populate_existing() .first() ) - if current_user_api is None: + # A caller who reached the gate through a personal link row and no + # longer has one gets the 404 that a caller with no association to + # this connector has always gotten. A caller who never had a personal + # row -- the stand-in -- legitimately has none, so its absence says + # nothing about that caller's authorization; the re-resolved team + # verdict below is what decides them. + if current_user_api is None and not is_stand_in: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Custom API not found", ) - if not current_user_api.can_edit: + + still_can_edit = current_user_api is not None and bool( + current_user_api.can_edit + ) + if not still_can_edit and team_access is not None and team_access.can_edit: + # The verdict that granted this edit was resolved before this + # lock existed, and the application answering it can revoke the + # link at any moment through its own tables, which this lock does + # not cover. Re-resolve it whenever it is the half of the gate's + # decision that still has to grant the edit; a fresh link row that + # grants edit on its own already settles the question and spends + # no further hook call. + # + # This re-check assumes READ COMMITTED, PostgreSQL's default, + # which this codebase sets no isolation_level on its engine to + # change: it needs a fresh snapshot to see a link the application + # revoked and committed after this request's pre-lock read. Under + # REPEATABLE READ or SERIALIZABLE the re-read reuses this + # transaction's original snapshot, sees the pre-lock answer again, + # and the re-check degrades to a no-op -- it would stop refusing, + # not start refusing wrongly. + from ..services.connector_team_scope import ( + resolve_one_connector_access_or_raise, + ) + + try: + rechecked = resolve_one_connector_access_or_raise( + db, int(current_user.id), ("custom_api", int(api_id)) + ) + except ConnectorRuntimeError as exc: + db.rollback() + raise _http_from_connector_runtime(exc) from exc + if rechecked is None or not rechecked.can_edit: + db.rollback() + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=( + "Your team's access to this Custom API changed while " + "this edit was in flight" + ), + ) + still_can_edit = True + + if not still_can_edit: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="You do not have permission to edit this Custom API", ) - user_api = current_user_api + if current_user_api is not None: + user_api = current_user_api # Read only after the statement above: rename_team_connector's "old" # argument must be the name this transaction's own read established -- - # under the lock, on the path that renames -- not whatever the - # relationship read further up saw. A concurrent committed rename in + # under the lock, on the path that renames -- not whatever the gate's + # read further up saw. A concurrent committed rename in # between would otherwise make this stale, and the rewrite below would # look for a name that no longer exists anywhere, leaving the previous # renamer's selectors dangling with no error. @@ -516,18 +696,28 @@ def update_custom_api( from ..services.connector_team_scope import rename_team_connector - rename_team_connector( - db, - int(current_user.id), - "custom_api", - int(api_id), - old_name, - str(api.name), - ) + # The rename hook is application-installed and raises the seam's own typed + # error; answer with the status that error declares rather than letting it + # reach the generic handler as a 500. Everything this request staged is + # rolled back first, so a refused rename leaves nothing behind. + try: + rename_team_connector( + db, + int(current_user.id), + "custom_api", + int(api_id), + old_name, + str(api.name), + ) + except ConnectorRuntimeError as exc: + db.rollback() + raise _http_from_connector_runtime(exc) from exc - # Update UserCustomApi link + # Update UserCustomApi link. A caller with no personal row never reaches + # this: a payload carrying is_active from such a caller was rejected by + # the guard above the lock. if api_data.is_active is not None: - user_api.is_active = api_data.is_active # type: ignore[assignment] + user_api.is_active = api_data.is_active db.commit() db.refresh(api) From 987c062caeb23ea36de6536052676c85b3ab271d Mon Sep 17 00:00:00 2001 From: Alexliu Date: Fri, 4 Sep 2026 01:26:35 +0800 Subject: [PATCH 02/10] test(connectors): cover team-shared Custom API editing Pins the resolution of a caller with no personal association row, the combined edit gate, the is_active rejection for such a caller, the typed error arm on every hook this route calls, the re-resolution of the team verdict under the definition row's lock, and how many hook calls each population of caller pays. --- .../test_custom_api_team_connector_edit.py | 646 ++++++++++++++++++ 1 file changed, 646 insertions(+) create mode 100644 tests/web/api/test_custom_api_team_connector_edit.py diff --git a/tests/web/api/test_custom_api_team_connector_edit.py b/tests/web/api/test_custom_api_team_connector_edit.py new file mode 100644 index 000000000..33ebcfcd9 --- /dev/null +++ b/tests/web/api/test_custom_api_team_connector_edit.py @@ -0,0 +1,646 @@ +"""The edit right on a team-linked Custom API: ``GET``/``PUT +/api/custom-apis/{api_id}`` resolve a caller with no personal row through +the connector access hook instead of 404ing outright, ``can_edit`` falls +back to that verdict for a caller with no personal row, an ``is_active`` +payload from such a caller rejects outright instead of writing a shadow +attribute the response then reads back, a raising hook surfaces as its +declared status rather than a 500, and the verdict is re-resolved once +more after the definition row's lock is taken, refusing the write if it +no longer grants what the pre-lock answer granted. + +Every test installs the access hook through ``snapshot_connector_team_hooks`` +so no hook state leaks between tests or into suites that run after this one. +""" + +from __future__ import annotations + +import pytest +from fastapi import HTTPException +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from xagent.core.tools.adapters.vibe.connector_runtime import ConnectorRuntimeError +from xagent.web.api.custom_api import ( + CustomApiUpdate, + get_custom_api, + update_custom_api, +) +from xagent.web.models.custom_api import CustomApi, UserCustomApi +from xagent.web.models.database import Base +from xagent.web.models.user import User +from xagent.web.services.connector_team_scope import ( + ConnectorAccess, + set_connector_team_hooks, + snapshot_connector_team_hooks, +) + + +@pytest.fixture() +def db(): + engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(engine) + session_factory = sessionmaker(autocommit=False, autoflush=False, bind=engine) + session = session_factory() + try: + yield session + finally: + session.close() + + +def _make_user(db, user_id: int, *, is_admin: bool = False) -> User: + user = User( + id=user_id, username=f"user-{user_id}", password_hash="x", is_admin=is_admin + ) + db.add(user) + db.commit() + return user + + +def _make_owned_api(db, owner_id: int, *, name: str = "shared-api") -> CustomApi: + api = CustomApi(name=name, url="https://example.test/api", method="GET") + db.add(api) + db.flush() + db.add( + UserCustomApi( + user_id=owner_id, + custom_api_id=api.id, + is_owner=True, + can_edit=True, + can_delete=True, + is_active=True, + ) + ) + db.commit() + return api + + +async def _get(api_id, current_user, db): + return get_custom_api(api_id, current_user=current_user, db=db) + + +async def _put(api_id, payload, current_user, db): + return update_custom_api(api_id, payload, current_user=current_user, db=db) + + +def _sequenced_access_hook(*answers): + """An access hook that answers differently on successive calls, so a + test can make the second (post-lock) resolution disagree with the + first. ``None`` in the sequence means an empty answer -- the batch + contract's way of saying "the caller's team does not link this". An + entry that is an exception instance is raised instead of returned, so a + test can make the second resolution fail outright. The last entry + repeats for any further call. Records every call's ``refs`` on + ``.calls`` so a test can pin how many round trips the route pays.""" + calls: list[object] = [] + + def hook(db, user_id, refs): + calls.append(refs) + index = min(len(calls) - 1, len(answers) - 1) + answer = answers[index] + if isinstance(answer, BaseException): + raise answer + if answer is None: + return {} + return {ref: answer for ref in refs} + + hook.calls = calls + return hook + + +class TestGateHelperOnGetAndPut: + @pytest.mark.asyncio + async def test_get_404s_for_an_unrelated_user_with_no_link_and_no_team_access( + self, db + ): + owner = _make_user(db, 1) + stranger = _make_user(db, 2) + api = _make_owned_api(db, owner.id) + + with snapshot_connector_team_hooks(): + set_connector_team_hooks(access=lambda db, user_id, refs: {}) + with pytest.raises(HTTPException) as exc: + await _get(api.id, stranger, db) + assert exc.value.status_code == 404 + + @pytest.mark.asyncio + async def test_get_returns_the_stand_in_for_a_team_member_with_no_personal_row( + self, db + ): + owner = _make_user(db, 1) + member = _make_user(db, 2) + api = _make_owned_api(db, owner.id) + + with snapshot_connector_team_hooks(): + set_connector_team_hooks( + access=lambda db, user_id, refs: { + ref: ConnectorAccess(team_owned=True, can_edit=True) for ref in refs + } + ) + response = await _get(api.id, member, db) + + assert response.id == api.id + assert response.user_id == member.id + + @pytest.mark.asyncio + async def test_get_owner_behaviour_is_unchanged_with_no_hook_installed(self, db): + owner = _make_user(db, 1) + api = _make_owned_api(db, owner.id) + + with snapshot_connector_team_hooks(): + set_connector_team_hooks() + response = await _get(api.id, owner, db) + + assert response.id == api.id + assert response.user_id == owner.id + + +class TestPutWiringForATeamEditor: + @pytest.mark.asyncio + async def test_team_editor_edit_is_durable_and_creates_no_association_row(self, db): + owner = _make_user(db, 1) + editor = _make_user(db, 2) + api = _make_owned_api(db, owner.id) + api_id = api.id + + with snapshot_connector_team_hooks(): + set_connector_team_hooks( + access=lambda db, user_id, refs: { + ref: ConnectorAccess(team_owned=True, can_edit=True) for ref in refs + } + ) + response = await _put( + api_id, + CustomApiUpdate(description="edited by the team"), + editor, + db, + ) + + assert response.description == "edited by the team" + + # Durability, not staging -- a same-session query would still see + # an uncommitted UPDATE even if the route never committed. + db.rollback() + refreshed = db.query(CustomApi).filter(CustomApi.id == api_id).one() + assert refreshed.description == "edited by the team" + + # The edit did not fabricate a personal association for the team + # editor -- that would be a get-or-create write on an + # authorization path. + assert ( + db.query(UserCustomApi).filter(UserCustomApi.user_id == editor.id).first() + is None + ) + + @pytest.mark.asyncio + async def test_a_member_with_a_personal_row_edits_the_shared_config_durably( + self, db + ): + """A caller whose own personal row does not grant edit, widened by a + granting team verdict. ``can_edit=False`` on the personal row is the + point -- it is what keeps ``_resolve_custom_api_for_request``'s + ``skip_resolution_when=lambda ua: bool(ua.can_edit)`` from + short-circuiting before the verdict is even resolved.""" + owner = _make_user(db, 1) + member = _make_user(db, 2) + api = _make_owned_api(db, owner.id, name="both-rows-custom-api") + api_id = api.id + db.add( + UserCustomApi( + user_id=member.id, + custom_api_id=api_id, + is_owner=False, + can_edit=False, + is_active=True, + ) + ) + db.commit() + + with snapshot_connector_team_hooks(): + set_connector_team_hooks( + access=lambda db, user_id, refs: { + ref: ConnectorAccess(team_owned=True, can_edit=True) for ref in refs + } + ) + response = await _put( + api_id, + CustomApiUpdate(description="widened-by-the-team"), + member, + db, + ) + + assert response.description == "widened-by-the-team" + + db.rollback() + refreshed = db.query(CustomApi).filter(CustomApi.id == api_id).one() + assert refreshed.description == "widened-by-the-team" + assert ( + db.query(UserCustomApi) + .filter( + UserCustomApi.user_id == member.id, + UserCustomApi.custom_api_id == api_id, + ) + .count() + == 1 + ) + + @pytest.mark.asyncio + async def test_view_only_team_member_cannot_tamper_the_shared_config(self, db): + owner = _make_user(db, 1) + member = _make_user(db, 2) + api = _make_owned_api(db, owner.id) + + with snapshot_connector_team_hooks(): + set_connector_team_hooks( + access=lambda db, user_id, refs: { + ref: ConnectorAccess(team_owned=True, can_edit=False) + for ref in refs + } + ) + with pytest.raises(HTTPException) as exc: + await _put( + api.id, + CustomApiUpdate(description="should not land"), + member, + db, + ) + assert exc.value.status_code == 403 + + +@pytest.mark.asyncio +async def test_a_denying_verdict_stand_in_is_403_on_an_empty_payload_too(db): + """The route's edit gate has no payload-shape carve-out: it requires the + edit right for every payload, including one that sets no field at all, so + a caller with no personal row whose team verdict denies edit is refused + before anything is read or written. Pinned here so the gate cannot be + narrowed to specific payload shapes without this failing. + """ + owner = _make_user(db, 1) + member = _make_user(db, 2) + api = _make_owned_api(db, owner.id, name="denying-stand-in-target") + api_id = api.id + # Captured as plain values, not read off ``api`` after the call: ``api`` + # and the ``refreshed`` row below share the same identity-mapped Python + # object in this session, so comparing one against the other after the + # call would be comparing the object with itself and could never fail. + original_name = str(api.name) + original_description = str(api.description) if api.description is not None else None + + with snapshot_connector_team_hooks(): + set_connector_team_hooks( + access=lambda db, user_id, refs: { + ref: ConnectorAccess(team_owned=True, can_edit=False) for ref in refs + } + ) + with pytest.raises(HTTPException) as exc: + await _put(api_id, CustomApiUpdate(), member, db) + assert exc.value.status_code == 403 + + db.rollback() + refreshed = db.query(CustomApi).filter(CustomApi.id == api_id).one() + assert refreshed.name == original_name + assert refreshed.description == original_description + assert ( + db.query(UserCustomApi).filter(UserCustomApi.user_id == member.id).count() == 0 + ) + + +class TestIsActiveRejectionForAStandIn: + @pytest.mark.asyncio + async def test_is_active_from_a_caller_with_no_personal_row_is_400_not_a_silent_drop( + self, db + ): + owner = _make_user(db, 1) + editor = _make_user(db, 2) + api = _make_owned_api(db, owner.id, name="unchanged-name") + api_id = api.id + + with snapshot_connector_team_hooks(): + set_connector_team_hooks( + access=lambda db, user_id, refs: { + ref: ConnectorAccess(team_owned=True, can_edit=True) for ref in refs + } + ) + with pytest.raises(HTTPException) as exc: + await _put( + api_id, + CustomApiUpdate(is_active=False), + editor, + db, + ) + + # 1. the declared status. + assert exc.value.status_code == 400 + assert "personal connection" in str(exc.value.detail) + + # 2. nothing persisted -- the exception was raised before any + # commit, so a same-session rollback-then-requery must still show + # no personal association row for this caller. + db.rollback() + assert ( + db.query(UserCustomApi).filter(UserCustomApi.user_id == editor.id).first() + is None + ) + + # 3. the response body does not claim the change -- the call + # raised rather than returning, so no ``CustomApiResponse`` ever + # left the route carrying an ``is_active`` value nothing wrote. + refreshed = db.query(CustomApi).filter(CustomApi.id == api_id).one() + assert refreshed.name == "unchanged-name" + + +class TestTypedErrorArm: + """A raising hook still surfaces its declared status for a caller with + no working personal row -- the verdict is genuinely the gate for that + population and must stay fail-closed. An owner's row already decides + ``GET``'s answer (it never reads the verdict at all) and ``PUT``'s + (``can_edit`` is already ``True``), so neither ever calls the hook for + an owner's row; that population is pinned separately, below, in + ``TestOwnerIsImmuneToAHookFailure``.""" + + @pytest.mark.asyncio + async def test_get_surfaces_a_raising_hooks_declared_status(self, db): + owner = _make_user(db, 1) + member = _make_user(db, 2) + api = _make_owned_api(db, owner.id) + + def boom(*_a, **_k): + raise ValueError("hook exploded") + + with snapshot_connector_team_hooks(): + set_connector_team_hooks(access=boom) + with pytest.raises(HTTPException) as exc: + await _get(api.id, member, db) + + assert exc.value.status_code == 503 + + @pytest.mark.asyncio + async def test_put_surfaces_a_raising_hooks_declared_status_and_leaves_the_row_unchanged( + self, db + ): + owner = _make_user(db, 1) + member = _make_user(db, 2) + api = _make_owned_api(db, owner.id, name="pristine") + api_id = api.id + + def boom(*_a, **_k): + raise ValueError("hook exploded") + + with snapshot_connector_team_hooks(): + set_connector_team_hooks(access=boom) + with pytest.raises(HTTPException) as exc: + await _put( + api_id, + CustomApiUpdate(name="should-not-land"), + member, + db, + ) + + assert exc.value.status_code == 503 + + db.rollback() + refreshed = db.query(CustomApi).filter(CustomApi.id == api_id).one() + assert refreshed.name == "pristine" + + @pytest.mark.asyncio + async def test_put_passes_through_a_planted_connector_runtime_error_by_its_own_status( + self, db + ): + owner = _make_user(db, 1) + member = _make_user(db, 2) + api = _make_owned_api(db, owner.id) + + def boom(*_a, **_k): + raise ConnectorRuntimeError("planted", "planted failure", status_code=409) + + with snapshot_connector_team_hooks(): + set_connector_team_hooks(access=boom) + with pytest.raises(HTTPException) as exc: + await _put( + api.id, + CustomApiUpdate(description="irrelevant"), + member, + db, + ) + + assert exc.value.status_code == 409 + assert exc.value.detail == "planted failure" + + @pytest.mark.asyncio + async def test_a_raising_rename_hook_surfaces_its_declared_status_not_a_500( + self, db + ): + """The rename hook runs after the definition row has already been + rewritten in the session. Its typed failure must reach the client as + the status the seam declares, not as a generic 500, and the staged + rename must not survive the refusal.""" + owner = _make_user(db, 1) + api = _make_owned_api(db, owner.id, name="rename-hook-raises") + api_id = api.id + original_name = str(api.name) + + def boom(*_a, **_k): + raise ConnectorRuntimeError( + "connector_runtime_unavailable", + "Connector team scope is unavailable.", + status_code=503, + ) + + with snapshot_connector_team_hooks(): + set_connector_team_hooks(renamed=boom) + with pytest.raises(HTTPException) as exc: + update_custom_api( + api_id, + CustomApiUpdate(name="renamed-by-the-test"), + current_user=owner, + db=db, + ) + + assert exc.value.status_code == 503 + # Zero side effects: the rename that triggered the hook is rolled + # back with everything else this request had staged. + db.rollback() + refreshed = db.query(CustomApi).filter(CustomApi.id == api_id).one() + assert refreshed.name == original_name + + +class TestOwnerIsImmuneToAHookFailure: + """An owner's row already decides both routes' answers on its own -- + ``GET`` never reads the verdict at all, and ``PUT``'s ``can_edit`` is + already ``True`` -- so neither ever calls the hook for an owner's row. + A hook that would raise must therefore never surface: both routes + return their normal success status, unaffected by whatever the hook + would have done.""" + + @pytest.mark.asyncio + async def test_get_and_put_succeed_for_an_owner_even_though_the_hook_would_raise( + self, db + ): + owner = _make_user(db, 1) + api = _make_owned_api(db, owner.id, name="owner-immune") + api_id = api.id + + def boom(*_a, **_k): + raise ValueError("hook exploded") + + with snapshot_connector_team_hooks(): + set_connector_team_hooks(access=boom) + get_response = await _get(api_id, owner, db) + put_response = await _put( + api_id, + CustomApiUpdate(description="edited by the owner"), + owner, + db, + ) + + assert get_response.id == api_id + assert put_response.description == "edited by the owner" + + +class TestTheVerdictIsRevalidatedUnderTheDefinitionLock: + """The verdict granting a caller edit access is resolved before the + route's definition-row lock exists, and the installing application can + revoke the link at any moment through its own tables, which that lock + does not cover. Every payload below writes the shared definition row, so + every one of them takes the lock and must therefore re-establish the + verdict under it before committing. + """ + + async def _run(self, db, *, hook): + owner = _make_user(db, 1) + member = _make_user(db, 2) + api = _make_owned_api(db, owner.id, name="revalidated-under-lock") + api_id = api.id + # Captured as plain values before the call, not read off ``api`` + # afterwards: ``api`` and the requery below share the same + # identity-mapped Python object in this session, so comparing one + # against the other after the call would be comparing the object + # with itself and could never fail. + original_name = str(api.name) + original_description = ( + str(api.description) if api.description is not None else None + ) + + with snapshot_connector_team_hooks(): + set_connector_team_hooks(access=hook) + result = {} + try: + result["response"] = await _put( + api_id, + CustomApiUpdate(description="edited-while-in-flight"), + member, + db, + ) + except HTTPException as exc: + result["error"] = exc + return api, api_id, result, original_name, original_description + + @pytest.mark.asyncio + async def test_revoked_between_resolution_and_lock_is_refused(self, db): + hook = _sequenced_access_hook( + ConnectorAccess(team_owned=True, can_edit=True), None + ) + _api, api_id, result, original_name, original_description = await self._run( + db, hook=hook + ) + + assert result["error"].status_code == 403 + db.rollback() + refreshed = db.query(CustomApi).filter(CustomApi.id == api_id).one() + assert refreshed.name == original_name + assert refreshed.description == original_description + assert db.query(UserCustomApi).filter(UserCustomApi.user_id == 2).count() == 0 + + @pytest.mark.asyncio + async def test_downgraded_to_not_editable_between_resolution_and_lock_is_refused( + self, db + ): + hook = _sequenced_access_hook( + ConnectorAccess(team_owned=True, can_edit=True), + ConnectorAccess(team_owned=True, can_edit=False), + ) + _api, api_id, result, original_name, original_description = await self._run( + db, hook=hook + ) + + assert result["error"].status_code == 403 + db.rollback() + refreshed = db.query(CustomApi).filter(CustomApi.id == api_id).one() + assert refreshed.name == original_name + assert refreshed.description == original_description + assert db.query(UserCustomApi).filter(UserCustomApi.user_id == 2).count() == 0 + + @pytest.mark.asyncio + async def test_still_granted_on_recheck_commits_durably(self, db): + hook = _sequenced_access_hook( + ConnectorAccess(team_owned=True, can_edit=True), + ConnectorAccess(team_owned=True, can_edit=True), + ) + ( + _api, + api_id, + result, + _original_name, + _original_description, + ) = await self._run(db, hook=hook) + + assert "error" not in result + assert result["response"].description == "edited-while-in-flight" + + db.rollback() + refreshed = db.query(CustomApi).filter(CustomApi.id == api_id).one() + assert refreshed.description == "edited-while-in-flight" + + @pytest.mark.asyncio + async def test_recheck_that_raises_surfaces_the_hooks_own_status_with_zero_side_effects( + self, db + ): + hook = _sequenced_access_hook( + ConnectorAccess(team_owned=True, can_edit=True), + ValueError("hook exploded during recheck"), + ) + _api, api_id, result, original_name, original_description = await self._run( + db, hook=hook + ) + + assert result["error"].status_code == 503 + db.rollback() + refreshed = db.query(CustomApi).filter(CustomApi.id == api_id).one() + assert refreshed.name == original_name + assert refreshed.description == original_description + assert db.query(UserCustomApi).filter(UserCustomApi.user_id == 2).count() == 0 + + +class TestTheRecheckCostsExactlyOneExtraHookCall: + """How many times a single request calls the application's access hook. + A caller admitted by a team verdict pays one call at the gate and one + more under the lock; a caller whose own link row already grants edit + pays none at all.""" + + @pytest.mark.asyncio + async def test_a_granting_stand_in_editing_the_shared_config_pays_two_calls( + self, db + ): + owner = _make_user(db, 1) + member = _make_user(db, 2) + api = _make_owned_api(db, owner.id, name="cost-stand-in-shared") + api_id = api.id + hook = _sequenced_access_hook(ConnectorAccess(team_owned=True, can_edit=True)) + + with snapshot_connector_team_hooks(): + set_connector_team_hooks(access=hook) + await _put(api_id, CustomApiUpdate(description="shared-edit"), member, db) + + assert len(hook.calls) == 2 + + @pytest.mark.asyncio + async def test_an_owner_pays_zero_calls(self, db): + owner = _make_user(db, 1) + api = _make_owned_api(db, owner.id, name="cost-owner") + api_id = api.id + hook = _sequenced_access_hook(ConnectorAccess(team_owned=True, can_edit=True)) + + with snapshot_connector_team_hooks(): + set_connector_team_hooks(access=hook) + await _put(api_id, CustomApiUpdate(description="owner-edit"), owner, db) + + assert len(hook.calls) == 0 From 4bea4f1efd9b601071086d507d0f886fbdc967e8 Mon Sep 17 00:00:00 2001 From: Alexliu Date: Fri, 4 Sep 2026 01:41:01 +0800 Subject: [PATCH 03/10] test(connectors): keep the connector seam off the event loop thread Two of this module's routes reach an installed connector team hook only through a helper, so a hand-written list of routes is the wrong shape for this invariant. Discover them by transitive reachability from the module's own imports instead, assert the discovered set equals a written-out literal so the check cannot pass by finding nothing, and assert every member is a plain def. Also drops the two await keywords on get_custom_api, which is now a plain def for the same reason: an installed hook may do database work, and a coroutine route would run it on the event loop thread. --- tests/web/api/test_custom_api.py | 92 +++++++++++++++++++++++++++++++- 1 file changed, 90 insertions(+), 2 deletions(-) diff --git a/tests/web/api/test_custom_api.py b/tests/web/api/test_custom_api.py index 97df7d46d..48a016501 100644 --- a/tests/web/api/test_custom_api.py +++ b/tests/web/api/test_custom_api.py @@ -1,4 +1,5 @@ import ast +import importlib import inspect from datetime import datetime from types import SimpleNamespace @@ -29,6 +30,93 @@ set_connector_team_hooks, ) +_SEAM_MODULE = "xagent.web.api.custom_api" + +# Every top-level function in this module that can reach an installed +# connector team hook. Written out so the discovery below cannot pass by +# finding nothing. +_SEAM_REACHING_FUNCTIONS = { + "_resolve_custom_api_for_request", + "get_custom_api", + "update_custom_api", + "delete_custom_api", +} + + +def _functions_reaching_the_connector_seam() -> dict[str, ast.AST]: + """Every top-level function in this module that can reach an installed + connector team hook. + + Seeded on the functions that import ``connector_team_scope`` in their own + body, which is how every call site in this module reaches the seam, then + closed transitively over plain-name calls, because one route reaches it + only through a helper (``get_custom_api`` through + ``_resolve_custom_api_for_request``). A seed-only check would miss exactly + the route this test exists for. + """ + module = importlib.import_module(_SEAM_MODULE) + tree = ast.parse(inspect.getsource(module)) + functions = { + node.name: node + for node in tree.body + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + } + reaching = { + name + for name, node in functions.items() + if any( + isinstance(child, ast.ImportFrom) + and child.module is not None + and child.module.endswith("connector_team_scope") + for child in ast.walk(node) + ) + } + changed = True + while changed: + changed = False + for name, node in functions.items(): + if name in reaching: + continue + called = { + child.func.id + for child in ast.walk(node) + if isinstance(child, ast.Call) and isinstance(child.func, ast.Name) + } + if called & reaching: + reaching.add(name) + changed = True + return {name: functions[name] for name in reaching} + + +def test_the_discovery_of_seam_reaching_functions_is_not_vacuous(): + """Pins the enumeration itself, so the assertion below cannot pass by + finding nothing.""" + assert set(_functions_reaching_the_connector_seam()) == _SEAM_REACHING_FUNCTIONS + + +def test_no_function_that_reaches_the_connector_seam_is_a_coroutine(): + """An installed connector team hook may be slow -- the seam is designed on + the assumption that the installing application answers from its own + tables. FastAPI runs a coroutine route on the event loop thread itself, so + a slow hook call inside an ``async def`` stalls every other request the + process is serving, not just this one; a plain ``def`` goes to the + threadpool instead, where a slow call occupies one worker. + + Enumerated by reachability rather than by a hand-written list of routes: + an earlier fix for this same risk class swept siblings along the "takes a + row lock" axis and therefore missed a route that calls a hook without + taking one. + """ + offenders = [ + name + for name, node in _functions_reaching_the_connector_seam().items() + if isinstance(node, ast.AsyncFunctionDef) + ] + assert offenders == [], ( + "these functions can reach an installed connector team hook while " + f"running on the event loop thread: {sorted(offenders)}" + ) + def test_custom_api_models_env_validation(): # Valid creation @@ -262,7 +350,7 @@ async def test_get_custom_api(): db.query().filter().first.return_value = mock_user_api - res = await get_custom_api(10, current_user=user, db=db) + res = get_custom_api(10, current_user=user, db=db) assert res.id == 10 assert res.name == "test_api" @@ -274,7 +362,7 @@ async def test_get_custom_api_not_found(): db.query().filter().first.return_value = None with pytest.raises(HTTPException) as exc_info: - await get_custom_api(99, current_user=user, db=db) + get_custom_api(99, current_user=user, db=db) assert exc_info.value.status_code == 404 From f450d7afd28ae466ac21ff0ee65aec7c18d1c77e Mon Sep 17 00:00:00 2001 From: Alexliu Date: Fri, 4 Sep 2026 09:13:40 +0800 Subject: [PATCH 04/10] test(connectors): call the synchronous Custom API routes synchronously get_custom_api and update_custom_api are plain defs so an installed connector team hook's database work stays off the event loop thread. This file still wrapped every call in async def and await, which reads as if the routes were coroutines and contradicts the plain-def invariant the suite now pins. Drop the coroutine wrappers, the redundant asyncio markers, and the awaits; the calls are direct. --- .../test_custom_api_team_connector_edit.py | 113 +++++++----------- 1 file changed, 42 insertions(+), 71 deletions(-) diff --git a/tests/web/api/test_custom_api_team_connector_edit.py b/tests/web/api/test_custom_api_team_connector_edit.py index 33ebcfcd9..9397a2cc6 100644 --- a/tests/web/api/test_custom_api_team_connector_edit.py +++ b/tests/web/api/test_custom_api_team_connector_edit.py @@ -74,11 +74,11 @@ def _make_owned_api(db, owner_id: int, *, name: str = "shared-api") -> CustomApi return api -async def _get(api_id, current_user, db): +def _get(api_id, current_user, db): return get_custom_api(api_id, current_user=current_user, db=db) -async def _put(api_id, payload, current_user, db): +def _put(api_id, payload, current_user, db): return update_custom_api(api_id, payload, current_user=current_user, db=db) @@ -108,10 +108,7 @@ def hook(db, user_id, refs): class TestGateHelperOnGetAndPut: - @pytest.mark.asyncio - async def test_get_404s_for_an_unrelated_user_with_no_link_and_no_team_access( - self, db - ): + def test_get_404s_for_an_unrelated_user_with_no_link_and_no_team_access(self, db): owner = _make_user(db, 1) stranger = _make_user(db, 2) api = _make_owned_api(db, owner.id) @@ -119,13 +116,10 @@ async def test_get_404s_for_an_unrelated_user_with_no_link_and_no_team_access( with snapshot_connector_team_hooks(): set_connector_team_hooks(access=lambda db, user_id, refs: {}) with pytest.raises(HTTPException) as exc: - await _get(api.id, stranger, db) + _get(api.id, stranger, db) assert exc.value.status_code == 404 - @pytest.mark.asyncio - async def test_get_returns_the_stand_in_for_a_team_member_with_no_personal_row( - self, db - ): + def test_get_returns_the_stand_in_for_a_team_member_with_no_personal_row(self, db): owner = _make_user(db, 1) member = _make_user(db, 2) api = _make_owned_api(db, owner.id) @@ -136,27 +130,25 @@ async def test_get_returns_the_stand_in_for_a_team_member_with_no_personal_row( ref: ConnectorAccess(team_owned=True, can_edit=True) for ref in refs } ) - response = await _get(api.id, member, db) + response = _get(api.id, member, db) assert response.id == api.id assert response.user_id == member.id - @pytest.mark.asyncio - async def test_get_owner_behaviour_is_unchanged_with_no_hook_installed(self, db): + def test_get_owner_behaviour_is_unchanged_with_no_hook_installed(self, db): owner = _make_user(db, 1) api = _make_owned_api(db, owner.id) with snapshot_connector_team_hooks(): set_connector_team_hooks() - response = await _get(api.id, owner, db) + response = _get(api.id, owner, db) assert response.id == api.id assert response.user_id == owner.id class TestPutWiringForATeamEditor: - @pytest.mark.asyncio - async def test_team_editor_edit_is_durable_and_creates_no_association_row(self, db): + def test_team_editor_edit_is_durable_and_creates_no_association_row(self, db): owner = _make_user(db, 1) editor = _make_user(db, 2) api = _make_owned_api(db, owner.id) @@ -168,7 +160,7 @@ async def test_team_editor_edit_is_durable_and_creates_no_association_row(self, ref: ConnectorAccess(team_owned=True, can_edit=True) for ref in refs } ) - response = await _put( + response = _put( api_id, CustomApiUpdate(description="edited by the team"), editor, @@ -191,10 +183,7 @@ async def test_team_editor_edit_is_durable_and_creates_no_association_row(self, is None ) - @pytest.mark.asyncio - async def test_a_member_with_a_personal_row_edits_the_shared_config_durably( - self, db - ): + def test_a_member_with_a_personal_row_edits_the_shared_config_durably(self, db): """A caller whose own personal row does not grant edit, widened by a granting team verdict. ``can_edit=False`` on the personal row is the point -- it is what keeps ``_resolve_custom_api_for_request``'s @@ -221,7 +210,7 @@ async def test_a_member_with_a_personal_row_edits_the_shared_config_durably( ref: ConnectorAccess(team_owned=True, can_edit=True) for ref in refs } ) - response = await _put( + response = _put( api_id, CustomApiUpdate(description="widened-by-the-team"), member, @@ -243,8 +232,7 @@ async def test_a_member_with_a_personal_row_edits_the_shared_config_durably( == 1 ) - @pytest.mark.asyncio - async def test_view_only_team_member_cannot_tamper_the_shared_config(self, db): + def test_view_only_team_member_cannot_tamper_the_shared_config(self, db): owner = _make_user(db, 1) member = _make_user(db, 2) api = _make_owned_api(db, owner.id) @@ -257,7 +245,7 @@ async def test_view_only_team_member_cannot_tamper_the_shared_config(self, db): } ) with pytest.raises(HTTPException) as exc: - await _put( + _put( api.id, CustomApiUpdate(description="should not land"), member, @@ -266,8 +254,7 @@ async def test_view_only_team_member_cannot_tamper_the_shared_config(self, db): assert exc.value.status_code == 403 -@pytest.mark.asyncio -async def test_a_denying_verdict_stand_in_is_403_on_an_empty_payload_too(db): +def test_a_denying_verdict_stand_in_is_403_on_an_empty_payload_too(db): """The route's edit gate has no payload-shape carve-out: it requires the edit right for every payload, including one that sets no field at all, so a caller with no personal row whose team verdict denies edit is refused @@ -292,7 +279,7 @@ async def test_a_denying_verdict_stand_in_is_403_on_an_empty_payload_too(db): } ) with pytest.raises(HTTPException) as exc: - await _put(api_id, CustomApiUpdate(), member, db) + _put(api_id, CustomApiUpdate(), member, db) assert exc.value.status_code == 403 db.rollback() @@ -305,8 +292,7 @@ async def test_a_denying_verdict_stand_in_is_403_on_an_empty_payload_too(db): class TestIsActiveRejectionForAStandIn: - @pytest.mark.asyncio - async def test_is_active_from_a_caller_with_no_personal_row_is_400_not_a_silent_drop( + def test_is_active_from_a_caller_with_no_personal_row_is_400_not_a_silent_drop( self, db ): owner = _make_user(db, 1) @@ -321,7 +307,7 @@ async def test_is_active_from_a_caller_with_no_personal_row_is_400_not_a_silent_ } ) with pytest.raises(HTTPException) as exc: - await _put( + _put( api_id, CustomApiUpdate(is_active=False), editor, @@ -357,8 +343,7 @@ class TestTypedErrorArm: an owner's row; that population is pinned separately, below, in ``TestOwnerIsImmuneToAHookFailure``.""" - @pytest.mark.asyncio - async def test_get_surfaces_a_raising_hooks_declared_status(self, db): + def test_get_surfaces_a_raising_hooks_declared_status(self, db): owner = _make_user(db, 1) member = _make_user(db, 2) api = _make_owned_api(db, owner.id) @@ -369,12 +354,11 @@ def boom(*_a, **_k): with snapshot_connector_team_hooks(): set_connector_team_hooks(access=boom) with pytest.raises(HTTPException) as exc: - await _get(api.id, member, db) + _get(api.id, member, db) assert exc.value.status_code == 503 - @pytest.mark.asyncio - async def test_put_surfaces_a_raising_hooks_declared_status_and_leaves_the_row_unchanged( + def test_put_surfaces_a_raising_hooks_declared_status_and_leaves_the_row_unchanged( self, db ): owner = _make_user(db, 1) @@ -388,7 +372,7 @@ def boom(*_a, **_k): with snapshot_connector_team_hooks(): set_connector_team_hooks(access=boom) with pytest.raises(HTTPException) as exc: - await _put( + _put( api_id, CustomApiUpdate(name="should-not-land"), member, @@ -401,8 +385,7 @@ def boom(*_a, **_k): refreshed = db.query(CustomApi).filter(CustomApi.id == api_id).one() assert refreshed.name == "pristine" - @pytest.mark.asyncio - async def test_put_passes_through_a_planted_connector_runtime_error_by_its_own_status( + def test_put_passes_through_a_planted_connector_runtime_error_by_its_own_status( self, db ): owner = _make_user(db, 1) @@ -415,7 +398,7 @@ def boom(*_a, **_k): with snapshot_connector_team_hooks(): set_connector_team_hooks(access=boom) with pytest.raises(HTTPException) as exc: - await _put( + _put( api.id, CustomApiUpdate(description="irrelevant"), member, @@ -425,10 +408,7 @@ def boom(*_a, **_k): assert exc.value.status_code == 409 assert exc.value.detail == "planted failure" - @pytest.mark.asyncio - async def test_a_raising_rename_hook_surfaces_its_declared_status_not_a_500( - self, db - ): + def test_a_raising_rename_hook_surfaces_its_declared_status_not_a_500(self, db): """The rename hook runs after the definition row has already been rewritten in the session. Its typed failure must reach the client as the status the seam declares, not as a generic 500, and the staged @@ -471,8 +451,7 @@ class TestOwnerIsImmuneToAHookFailure: return their normal success status, unaffected by whatever the hook would have done.""" - @pytest.mark.asyncio - async def test_get_and_put_succeed_for_an_owner_even_though_the_hook_would_raise( + def test_get_and_put_succeed_for_an_owner_even_though_the_hook_would_raise( self, db ): owner = _make_user(db, 1) @@ -484,8 +463,8 @@ def boom(*_a, **_k): with snapshot_connector_team_hooks(): set_connector_team_hooks(access=boom) - get_response = await _get(api_id, owner, db) - put_response = await _put( + get_response = _get(api_id, owner, db) + put_response = _put( api_id, CustomApiUpdate(description="edited by the owner"), owner, @@ -505,7 +484,7 @@ class TestTheVerdictIsRevalidatedUnderTheDefinitionLock: verdict under it before committing. """ - async def _run(self, db, *, hook): + def _run(self, db, *, hook): owner = _make_user(db, 1) member = _make_user(db, 2) api = _make_owned_api(db, owner.id, name="revalidated-under-lock") @@ -524,7 +503,7 @@ async def _run(self, db, *, hook): set_connector_team_hooks(access=hook) result = {} try: - result["response"] = await _put( + result["response"] = _put( api_id, CustomApiUpdate(description="edited-while-in-flight"), member, @@ -534,12 +513,11 @@ async def _run(self, db, *, hook): result["error"] = exc return api, api_id, result, original_name, original_description - @pytest.mark.asyncio - async def test_revoked_between_resolution_and_lock_is_refused(self, db): + def test_revoked_between_resolution_and_lock_is_refused(self, db): hook = _sequenced_access_hook( ConnectorAccess(team_owned=True, can_edit=True), None ) - _api, api_id, result, original_name, original_description = await self._run( + _api, api_id, result, original_name, original_description = self._run( db, hook=hook ) @@ -550,15 +528,14 @@ async def test_revoked_between_resolution_and_lock_is_refused(self, db): assert refreshed.description == original_description assert db.query(UserCustomApi).filter(UserCustomApi.user_id == 2).count() == 0 - @pytest.mark.asyncio - async def test_downgraded_to_not_editable_between_resolution_and_lock_is_refused( + def test_downgraded_to_not_editable_between_resolution_and_lock_is_refused( self, db ): hook = _sequenced_access_hook( ConnectorAccess(team_owned=True, can_edit=True), ConnectorAccess(team_owned=True, can_edit=False), ) - _api, api_id, result, original_name, original_description = await self._run( + _api, api_id, result, original_name, original_description = self._run( db, hook=hook ) @@ -569,8 +546,7 @@ async def test_downgraded_to_not_editable_between_resolution_and_lock_is_refused assert refreshed.description == original_description assert db.query(UserCustomApi).filter(UserCustomApi.user_id == 2).count() == 0 - @pytest.mark.asyncio - async def test_still_granted_on_recheck_commits_durably(self, db): + def test_still_granted_on_recheck_commits_durably(self, db): hook = _sequenced_access_hook( ConnectorAccess(team_owned=True, can_edit=True), ConnectorAccess(team_owned=True, can_edit=True), @@ -581,7 +557,7 @@ async def test_still_granted_on_recheck_commits_durably(self, db): result, _original_name, _original_description, - ) = await self._run(db, hook=hook) + ) = self._run(db, hook=hook) assert "error" not in result assert result["response"].description == "edited-while-in-flight" @@ -590,15 +566,14 @@ async def test_still_granted_on_recheck_commits_durably(self, db): refreshed = db.query(CustomApi).filter(CustomApi.id == api_id).one() assert refreshed.description == "edited-while-in-flight" - @pytest.mark.asyncio - async def test_recheck_that_raises_surfaces_the_hooks_own_status_with_zero_side_effects( + def test_recheck_that_raises_surfaces_the_hooks_own_status_with_zero_side_effects( self, db ): hook = _sequenced_access_hook( ConnectorAccess(team_owned=True, can_edit=True), ValueError("hook exploded during recheck"), ) - _api, api_id, result, original_name, original_description = await self._run( + _api, api_id, result, original_name, original_description = self._run( db, hook=hook ) @@ -616,10 +591,7 @@ class TestTheRecheckCostsExactlyOneExtraHookCall: more under the lock; a caller whose own link row already grants edit pays none at all.""" - @pytest.mark.asyncio - async def test_a_granting_stand_in_editing_the_shared_config_pays_two_calls( - self, db - ): + def test_a_granting_stand_in_editing_the_shared_config_pays_two_calls(self, db): owner = _make_user(db, 1) member = _make_user(db, 2) api = _make_owned_api(db, owner.id, name="cost-stand-in-shared") @@ -628,12 +600,11 @@ async def test_a_granting_stand_in_editing_the_shared_config_pays_two_calls( with snapshot_connector_team_hooks(): set_connector_team_hooks(access=hook) - await _put(api_id, CustomApiUpdate(description="shared-edit"), member, db) + _put(api_id, CustomApiUpdate(description="shared-edit"), member, db) assert len(hook.calls) == 2 - @pytest.mark.asyncio - async def test_an_owner_pays_zero_calls(self, db): + def test_an_owner_pays_zero_calls(self, db): owner = _make_user(db, 1) api = _make_owned_api(db, owner.id, name="cost-owner") api_id = api.id @@ -641,6 +612,6 @@ async def test_an_owner_pays_zero_calls(self, db): with snapshot_connector_team_hooks(): set_connector_team_hooks(access=hook) - await _put(api_id, CustomApiUpdate(description="owner-edit"), owner, db) + _put(api_id, CustomApiUpdate(description="owner-edit"), owner, db) assert len(hook.calls) == 0 From c36b9abf2d6b0d9ef778fcb1d9c9651b9fa52b60 Mon Sep 17 00:00:00 2001 From: Alexliu Date: Fri, 4 Sep 2026 20:42:45 +0800 Subject: [PATCH 05/10] test(connectors): pin the stand-in flags and the lock-free edit path Document in the schema what is_active and is_default mean for a caller with no personal link row to a Custom API: the response fields come from the _TeamOwnedUserApi stand-in's class constants rather than a stored column, and the same constants back the aggregate connector list's response for that caller. The request-side descriptions of is_active are brought in line with that wording, so both directions name the caller's own link row rather than "the API". Pin the equality between the two response surfaces with a new test rather than leaving it an implicit assumption shared by two response constructors. Add the missing success-path test for a team-granted stand-in caller submitting an empty payload: it writes no field of the shared definition row, so it never takes the row's lock and never re-resolves the edit verdict a second time, unlike every other case this suite covers. Bring the two GET tests this PR touches back in sync with the route under test, which is not async: drop the redundant asyncio marker and the async def on test_get_custom_api and test_get_custom_api_not_found. The six other async def tests the review names exercise `update_custom_api` and `delete_custom_api`; they predate this PR and are outside its change surface, so they are left as-is. The remaining four exercise `list_custom_apis` and `create_custom_api`, which are still `async def`. The MCP-side counterpart of the same stand-in shape (mcp.py's MCPServerResponse.is_active/is_default, and its own granting-verdict empty payload path) is out of scope here and tracked separately. --- src/xagent/web/api/custom_api.py | 40 +++++++-- tests/web/api/test_custom_api.py | 6 +- .../test_custom_api_team_connector_edit.py | 90 +++++++++++++++++++ 3 files changed, 127 insertions(+), 9 deletions(-) diff --git a/src/xagent/web/api/custom_api.py b/src/xagent/web/api/custom_api.py index 97440e339..d8df93719 100644 --- a/src/xagent/web/api/custom_api.py +++ b/src/xagent/web/api/custom_api.py @@ -54,7 +54,9 @@ class CustomApiCreate(BaseModel): allow_delegated_authorization: bool = Field( False, description="Allow runtime Authorization header binding" ) - is_active: bool = Field(True, description="Whether the API is active") + is_active: bool = Field( + True, description="Whether this caller's own link row to the API is active" + ) class CustomApiUpdate(BaseModel): @@ -82,7 +84,9 @@ class CustomApiUpdate(BaseModel): allow_delegated_authorization: Optional[bool] = Field( None, description="Allow runtime Authorization header binding" ) - is_active: Optional[bool] = Field(None, description="Whether the API is active") + is_active: Optional[bool] = Field( + None, description="Whether this caller's own link row to the API is active" + ) class CustomApiResponse(BaseModel): @@ -100,8 +104,25 @@ class CustomApiResponse(BaseModel): runtime_input_schema: Optional[Dict[str, Any]] runtime_bindings: Optional[List[Dict[str, Any]]] allow_delegated_authorization: bool - is_active: bool - is_default: bool + is_active: bool = Field( + ..., + description=( + "Whether this caller's own link row to the API is active. A " + "caller who has no personal link row -- a team member reaching " + "a connector their team links -- has no row to hold this, and " + "receives the stand-in association's constant instead of a " + "stored value." + ), + ) + is_default: bool = Field( + ..., + description=( + "Whether this caller's own link row marks the API as their " + "default. Carries the same caveat as ``is_active``: with no " + "personal link row the value is the stand-in association's " + "constant, not a stored one." + ), + ) created_at: str updated_at: str @@ -118,7 +139,16 @@ def _db_api_to_response( api: CustomApi, user_api: "UserCustomApi | _TeamOwnedUserApi", ) -> CustomApiResponse: - """Convert database CustomApi to response model with masked env values.""" + """Convert database CustomApi to response model with masked env values. + + ``is_active`` and ``is_default`` are read off ``user_api``, which is the + caller's own link row when one exists and the ``_TeamOwnedUserApi`` + stand-in when the caller has no usable one. The stand-in holds class + constants rather than stored columns, so for such a caller these two + fields report those constants -- the same constants the aggregate + connector list carries for that caller whenever it lists the connector + at all, since it builds its response from the same stand-in. + """ # Mask env values for frontend masked_env = None diff --git a/tests/web/api/test_custom_api.py b/tests/web/api/test_custom_api.py index 48a016501..79bf54bc3 100644 --- a/tests/web/api/test_custom_api.py +++ b/tests/web/api/test_custom_api.py @@ -332,8 +332,7 @@ async def test_create_custom_api_rejects_runtime_static_header_conflict(): assert "Invalid runtime configuration" in str(exc_info.value.detail) -@pytest.mark.asyncio -async def test_get_custom_api(): +def test_get_custom_api(): db = MagicMock(spec=Session) user = User(id=1) @@ -355,8 +354,7 @@ async def test_get_custom_api(): assert res.name == "test_api" -@pytest.mark.asyncio -async def test_get_custom_api_not_found(): +def test_get_custom_api_not_found(): db = MagicMock(spec=Session) user = User(id=1) db.query().filter().first.return_value = None diff --git a/tests/web/api/test_custom_api_team_connector_edit.py b/tests/web/api/test_custom_api_team_connector_edit.py index 9397a2cc6..5fc798e60 100644 --- a/tests/web/api/test_custom_api_team_connector_edit.py +++ b/tests/web/api/test_custom_api_team_connector_edit.py @@ -25,6 +25,7 @@ get_custom_api, update_custom_api, ) +from xagent.web.api.mcp import _custom_api_to_mcp_response, _TeamOwnedUserApi from xagent.web.models.custom_api import CustomApi, UserCustomApi from xagent.web.models.database import Base from xagent.web.models.user import User @@ -291,6 +292,56 @@ def test_a_denying_verdict_stand_in_is_403_on_an_empty_payload_too(db): ) +def test_a_granting_verdict_stand_in_succeeds_on_an_empty_payload_without_a_recheck( + db, +): + """The granting counterpart of the test above, and the one path that + reaches a 200 while skipping both the definition row's lock and the + post-lock re-authorization: an empty payload writes no field of the + shared definition row, so it takes no lock, and having taken no lock it + has no unbounded wait to re-establish the gate's decision across. + + The hook is sequenced to grant on its first answer and to deny on every + later one. If the route ever re-resolved the verdict on this payload, the + second answer would refuse the request with a 403; the request returning + normally, together with the single recorded hook call, is what pins the + skip. + """ + owner = _make_user(db, 1) + member = _make_user(db, 2) + api = _make_owned_api(db, owner.id, name="granting-stand-in-target") + api_id = api.id + original_name = str(api.name) + original_description = str(api.description) if api.description is not None else None + + hook = _sequenced_access_hook( + ConnectorAccess(team_owned=True, can_edit=True), + None, + ) + with snapshot_connector_team_hooks(): + set_connector_team_hooks(access=hook) + response = _put(api_id, CustomApiUpdate(), member, db) + + # 1. one resolution, not two: the post-lock re-authorization never ran. + assert len(hook.calls) == 1 + + # 2. the shared definition row is untouched -- read back from the + # database after a rollback rather than off the in-session object. + db.rollback() + refreshed = db.query(CustomApi).filter(CustomApi.id == api_id).one() + assert refreshed.name == original_name + assert refreshed.description == original_description + + # 3. no personal link row was created for a caller who had none. + assert ( + db.query(UserCustomApi).filter(UserCustomApi.user_id == member.id).count() == 0 + ) + + # 4. the response is the stand-in's own view of the connector. + assert response.id == api_id + assert response.user_id == member.id + + class TestIsActiveRejectionForAStandIn: def test_is_active_from_a_caller_with_no_personal_row_is_400_not_a_silent_drop( self, db @@ -334,6 +385,45 @@ def test_is_active_from_a_caller_with_no_personal_row_is_400_not_a_silent_drop( assert refreshed.name == "unchanged-name" +class TestStandInFlagsAgreeAcrossBothResponseSurfaces: + def test_a_stand_in_caller_gets_the_same_two_flags_from_both_surfaces(self, db): + """``is_active`` and ``is_default`` live on the caller's own link row. + A caller with no such row is answered from a stand-in association + holding class constants, and two separate response constructors read + it: the single-connector ``GET`` here, and the aggregate connector + list's ``_custom_api_to_mcp_response``. Whatever those constants are, + both surfaces must report the same pair for the same caller -- pinned + as an equality rather than as literal ``True``/``False`` so that + changing what a stand-in reports stays a one-line change in one place + instead of a test failure that reads like a regression. + """ + owner = _make_user(db, 1) + member = _make_user(db, 2) + api = _make_owned_api(db, owner.id, name="two-surface-target") + api_id = api.id + + with snapshot_connector_team_hooks(): + set_connector_team_hooks( + access=lambda db, user_id, refs: { + ref: ConnectorAccess(team_owned=True, can_edit=False) + for ref in refs + } + ) + detail = _get(api_id, member, db) + + definition = db.query(CustomApi).filter(CustomApi.id == api_id).one() + aggregate = _custom_api_to_mcp_response( + definition, _TeamOwnedUserApi(int(member.id)) + ) + + assert ( + db.query(UserCustomApi).filter(UserCustomApi.user_id == member.id).first() + is None + ), "the caller under test must have no personal link row" + assert detail.is_active == aggregate.is_active + assert detail.is_default == aggregate.is_default + + class TestTypedErrorArm: """A raising hook still surfaces its declared status for a caller with no working personal row -- the verdict is genuinely the gate for that From af3020b607c536f2631d556591ee18ec7d3f8c4a Mon Sep 17 00:00:00 2001 From: Alexliu Date: Sun, 6 Sep 2026 22:52:05 +0800 Subject: [PATCH 06/10] fix(connectors): re-authorize under the lock through the team decision update_custom_api's post-lock re-authorization only re-resolved the team access verdict when the caller had reached the pre-lock gate through that verdict. A caller who reached the gate through a personal link row that was deleted, or downgraded, while the request waited for the definition row's lock fell through a 404 branch instead, without ever asking whether the team still grants the edit. Widen the re-check to run whenever the freshly-read personal row does not grant the edit on its own, replace the stale 404 branch with a four-way decision (grant, revoked-team-access, no-surviving-row, no-permission), and construct a fresh stand-in object when no personal row survives rather than carrying forward one that may reference a row deleted mid-wait. Add a second is_active guard under the lock for the same reason the pre-lock guard exists: a personal row that existed when the gate ran can be gone by the time the lock is taken. The re-check's hook call moves into its own function, _recheck_team_access_under_definition_lock, and declares caller_holds_lock=True: it runs after this route has taken the definition row FOR UPDATE and before anything staged is committed, so a hook that ends the transaction would release that lock unnoticed. Keeping it out of update_custom_api's body also keeps that route's hook calls one-per-function, which is how the call-site table in connector_team_scope.py is keyed; the table gains rows for this new function and for the pre-lock resolution helper, which reaches the same hook without holding any lock. --- src/xagent/web/api/custom_api.py | 174 +++++++++++++----- .../web/services/connector_team_scope.py | 12 +- tests/web/api/test_custom_api.py | 1 + 3 files changed, 134 insertions(+), 53 deletions(-) diff --git a/src/xagent/web/api/custom_api.py b/src/xagent/web/api/custom_api.py index 632b46e2a..04c589cbc 100644 --- a/src/xagent/web/api/custom_api.py +++ b/src/xagent/web/api/custom_api.py @@ -298,11 +298,13 @@ def _http_from_connector_runtime(exc: ConnectorRuntimeError) -> HTTPException: """Map the connector team seam's typed error onto this module's HTTP answer. Four call sites need it -- ``get_custom_api``, and ``update_custom_api`` - for its pre-lock resolution, its post-lock re-check and its rename hook -- - so that every one of them answers with the status and message the seam - declares instead of letting the error reach the generic handler as a 500. - Each call site wraps only the seam call it makes, rather than the whole - route body, so the mapping itself lives here in one place. + for its pre-lock resolution, its post-lock re-check (which makes its raw + hook call through ``_recheck_team_access_under_definition_lock``) and its rename + hook -- so that every one of them answers with the status and message + the seam declares instead of letting the error reach the generic + handler as a 500. Each call site wraps only the seam call it makes, + rather than the whole route body, so the mapping itself lives here in + one place. """ return HTTPException(status_code=exc.status_code, detail=exc.safe_message) @@ -387,6 +389,40 @@ def _resolve_custom_api_for_request( return resolved_user_api, cast(CustomApi, api), access +def _recheck_team_access_under_definition_lock( + db: Session, user_id: int, api_id: int +) -> "ConnectorAccess | None": + """Re-resolve the caller's team access verdict for ``update_custom_api``, + on behalf of a caller whose freshly-read personal link row does not + grant the edit on its own. + + Lives as its own function, separate from ``update_custom_api``, so that + this call site and the route's rename hook call are each attributed to a + distinct function by the call-site table's accounting + (``connector_team_scope.py``'s "Call sites and what the caller holds" + table, checked against the source by + ``test_the_call_site_table_and_the_call_sites_agree``): that accounting + is keyed by enclosing function name, one row per function, and + ``update_custom_api`` already owns the rename hook's row. + + Called only from inside ``update_custom_api``'s ``FOR UPDATE`` block, on + a payload that writes the ``custom_apis`` definition row, after that + lock has been taken and before anything this request has staged is + committed. Declares ``caller_holds_lock=True`` on the call it makes: a + hook that ends this transaction would release that lock without the + caller finding out, and the writes staged above would then commit + against a row somebody else may have moved. + + Raises ``ConnectorRuntimeError`` when access resolution itself fails; + the caller translates that into an ``HTTPException`` and rolls back. + """ + from ..services.connector_team_scope import resolve_one_connector_access_or_raise + + return resolve_one_connector_access_or_raise( + db, user_id, ("custom_api", api_id), caller_holds_lock=True + ) + + @custom_api_router.get("/{api_id}", response_model=CustomApiResponse) def get_custom_api( api_id: int, @@ -555,11 +591,12 @@ def update_custom_api( # # So both halves of the gate's decision are re-established here from # fresh reads and the same combination is recomputed: the caller's - # own link row is re-read, and the team verdict is re-resolved when - # it is the half that has to grant the edit. A payload that writes - # only the caller's own link row skips this block entirely, and - # correctly so -- it took no lock above, so its gate was never - # separated from its writes by a wait of unbounded length. + # own link row is re-read, and the team verdict is re-resolved + # whenever that fresh read does not grant the edit on its own. A + # payload that writes only the caller's own link row skips this + # block entirely, and correctly so -- it took no lock above, so its + # gate was never separated from its writes by a wait of unbounded + # length. # # ``populate_existing()`` on the definition query above refreshes # the row of that statement and nothing else, so it does not cover @@ -568,9 +605,12 @@ def update_custom_api( # ``populate_existing()`` so its columns are overwritten with the # database's current values rather than the ones the gate loaded. # When it returns a row, that row replaces ``user_api`` for the rest - # of the route -- the link-row write below and the response both read - # it; when it returns nothing, the caller had no personal row to begin - # with and keeps the stand-in it was resolved with. + # of the route -- the link-row write below and the response both + # read it. When it returns nothing, ``user_api`` becomes a freshly + # constructed stand-in rather than the one this route resolved + # before the lock: that earlier one may itself have been backed by + # a personal row that has since been deleted, and this statement is + # what finds that out. # A revocation that commits after these statements is the window # this route had before it took any lock at all: in-process, with # no wait in it. Closing that one needs an authorization fence of @@ -584,29 +624,27 @@ def update_custom_api( .populate_existing() .first() ) - # A caller who reached the gate through a personal link row and no - # longer has one gets the 404 that a caller with no association to - # this connector has always gotten. A caller who never had a personal - # row -- the stand-in -- legitimately has none, so its absence says - # nothing about that caller's authorization; the re-resolved team - # verdict below is what decides them. - if current_user_api is None and not is_stand_in: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail="Custom API not found", - ) - still_can_edit = current_user_api is not None and bool( current_user_api.can_edit ) - if not still_can_edit and team_access is not None and team_access.can_edit: - # The verdict that granted this edit was resolved before this - # lock existed, and the application answering it can revoke the - # link at any moment through its own tables, which this lock does - # not cover. Re-resolve it whenever it is the half of the gate's - # decision that still has to grant the edit; a fresh link row that - # grants edit on its own already settles the question and spends - # no further hook call. + if not still_can_edit: + # The freshly-read personal row does not grant the edit on its + # own -- either it is gone, or its can_edit has been cleared -- + # so the team verdict is re-resolved regardless of what the gate + # decided before the lock: unlike the narrower condition this + # replaces, this one also catches a caller whose personal row + # was deleted out from under a gate decision that never went + # through the team verdict at all. An owner (the fresh row still + # grants edit on its own) and a payload that writes only + # ``is_active`` (this whole block is skipped) both still cost + # nothing extra; a caller who reached the gate on the team + # verdict and still holds it spends the one extra hook call it + # always did; a request with no access hook installed spends + # nothing, because the hook slot being empty makes resolution + # return ``{}`` without a query. Only a request whose personal + # row or team verdict actually changed while it waited for the + # lock now gets a different -- and correct -- answer than it did + # before this change. # # This re-check assumes READ COMMITTED, PostgreSQL's default, # which this codebase sets no isolation_level on its engine to @@ -616,18 +654,19 @@ def update_custom_api( # transaction's original snapshot, sees the pre-lock answer again, # and the re-check degrades to a no-op -- it would stop refusing, # not start refusing wrongly. - from ..services.connector_team_scope import ( - resolve_one_connector_access_or_raise, - ) - try: - rechecked = resolve_one_connector_access_or_raise( - db, int(current_user.id), ("custom_api", int(api_id)) + rechecked = _recheck_team_access_under_definition_lock( + db, int(current_user.id), api_id ) except ConnectorRuntimeError as exc: db.rollback() raise _http_from_connector_runtime(exc) from exc - if rechecked is None or not rechecked.can_edit: + if rechecked is not None and rechecked.can_edit: + still_can_edit = True + elif team_access is not None and team_access.can_edit: + # The gate's own read once granted this edit through the + # team verdict; that access was revoked while this request + # waited for the lock. db.rollback() raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, @@ -636,15 +675,51 @@ def update_custom_api( "this edit was in flight" ), ) - still_can_edit = True + elif current_user_api is None: + # No personal row survived the wait, and neither the fresh + # nor the pre-lock team verdict grants the edit: the same + # 404 a caller with no association to this connector has + # always gotten, whether that caller reached the gate + # through a personal row that is now gone or never had one. + db.rollback() + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Custom API not found", + ) + else: + db.rollback() + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="You do not have permission to edit this Custom API", + ) - if not still_can_edit: - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail="You do not have permission to edit this Custom API", - ) if current_user_api is not None: user_api = current_user_api + else: + # The gate's own read may have resolved a personal row that has + # since been deleted; carrying that ORM object forward would + # raise ``StaleDataError`` from ``db.commit()`` below on a + # payload that writes the definition row, or + # ``ObjectDeletedError`` while the response is built on a + # payload that does not -- and the latter fails only after this + # request's other writes have already committed. A freshly + # constructed stand-in never touches the database at all, so + # neither can happen. + from .mcp import _TeamOwnedUserApi + + user_api = _TeamOwnedUserApi(int(current_user.id)) + + # A second is_active guard, under the lock. The one above the lock + # (see the ``is_stand_in`` check earlier in this route) catches a + # caller who already had no personal row when the gate ran; this one + # catches a caller whose personal row existed then and was deleted + # while this request waited for the lock. + if current_user_api is None and api_data.is_active is not None: + db.rollback() + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="No personal connection exists to configure is_active for this API", + ) # Read only after the statement above: rename_team_connector's "old" # argument must be the name this transaction's own read established -- @@ -749,8 +824,11 @@ def update_custom_api( raise _http_from_connector_runtime(exc) from exc # Update UserCustomApi link. A caller with no personal row never reaches - # this: a payload carrying is_active from such a caller was rejected by - # the guard above the lock. + # this: a payload carrying is_active is rejected by one of two guards -- + # the one above the lock, for a caller who already had no personal row + # when the gate ran, or the one under the lock, for a caller whose + # personal row existed then and was deleted while this request waited + # for the lock. if api_data.is_active is not None: user_api.is_active = api_data.is_active diff --git a/src/xagent/web/services/connector_team_scope.py b/src/xagent/web/services/connector_team_scope.py index eabf1cb10..0a15e73ad 100644 --- a/src/xagent/web/services/connector_team_scope.py +++ b/src/xagent/web/services/connector_team_scope.py @@ -47,10 +47,12 @@ | Call site | Caller holds while the hook runs | Committed before asking | ``caller_holds_lock`` | | --- | --- | --- | --- | | ``custom_api.update_custom_api`` | the ``custom_apis`` definition row, ``FOR UPDATE``, on the payloads that write that row | no | ``True`` | +| ``custom_api._recheck_team_access_under_definition_lock`` | the ``custom_apis`` definition row, ``FOR UPDATE``, taken by ``update_custom_api`` before this call | no | ``True`` | | ``custom_api.delete_custom_api`` | the ``custom_apis`` definition row, ``FOR UPDATE`` | no | ``True`` | | ``mcp.update_mcp_server`` | the ``mcp_servers`` definition row, ``FOR UPDATE ... KEY SHARE``, on the payloads that write that row | no | ``True`` | | ``mcp.teardown_mcp_app_server`` | three row locks: ``public_mcp_apps``, ``mcp_servers``, ``user_mcpservers`` | no, within this function -- see the note below | ``True`` | | ``mcp.delete_mcp_server`` | two row locks: ``mcp_servers`` and ``user_mcpservers``, taken by ``_lock_active_mcp_oauth_lifecycle`` before this call | no | ``True`` | +| ``custom_api._resolve_custom_api_for_request`` | nothing -- this resolution runs before either of its two routes takes any lock | no | ``False`` | ``mcp.teardown_mcp_app_server`` is a helper, not a route: it has no route decorator and no caller in this repository outside tests. "Nothing committed @@ -69,11 +71,11 @@ The remaining slots declare nothing. Every ``visibility`` and ``team_visibility`` call site is lock-free, and one of the ``team_visibility`` paths runs on a lazily created session that may not be in a transaction at -all. ``access`` has no call site in this repository; a caller that adds one -while holding a lock owes this table a row and owes the call -``caller_holds_lock=True``. One shape must never declare it: a call site that -has already committed its own work before asking, because refusing there -reports a failure for an operation that fully succeeded. +all. A caller that adds a new ``access`` call site while holding a lock owes +this table a row and owes the call ``caller_holds_lock=True``. One shape must +never declare it: a call site that has already committed its own work before +asking, because refusing there reports a failure for an operation that fully +succeeded. What the check is not --------------------- diff --git a/tests/web/api/test_custom_api.py b/tests/web/api/test_custom_api.py index 79bf54bc3..ba56fd5d6 100644 --- a/tests/web/api/test_custom_api.py +++ b/tests/web/api/test_custom_api.py @@ -37,6 +37,7 @@ # finding nothing. _SEAM_REACHING_FUNCTIONS = { "_resolve_custom_api_for_request", + "_recheck_team_access_under_definition_lock", "get_custom_api", "update_custom_api", "delete_custom_api", From 070b3bc2cd9c6420d5aa8262d9894c067ebf46de Mon Sep 17 00:00:00 2001 From: Alexliu Date: Sun, 6 Sep 2026 22:52:27 +0800 Subject: [PATCH 07/10] test(connectors): pin the post-lock re-authorization on both backends Covers update_custom_api's widened post-lock re-check and its four-way decision (grant, revoked-team-access, no-surviving-row, no-permission) plus the new locked is_active guard and stand-in construction. The SQLite-backed suite pins the caller_holds_lock declaration (a hook that ends the transaction on the post-lock call is refused as a boundary violation, not answered as if the lock still held), the hook-call-count contract across the four payload shapes that decide it (owner, is_active only, stand-in shared write, no hook installed), and the reachable form of the path where a can_edit=False personal row writes is_active on the strength of the team verdict without ever taking the lock. FOR UPDATE is a no-op on SQLite, so the interleavings where a personal row is deleted or downgraded while the lock is held -- and where the team verdict does or does not still grant the edit once re-asked -- need a real second connection to construct; those six cases go in the PostgreSQL-only suite, reusing its existing revoke-after-lock fixture. --- .../test_custom_api_edit_lock_postgresql.py | 356 ++++++++++++++++++ .../test_custom_api_team_connector_edit.py | 159 ++++++++ .../web/services/test_connector_team_scope.py | 8 +- 3 files changed, 520 insertions(+), 3 deletions(-) diff --git a/tests/web/api/test_custom_api_edit_lock_postgresql.py b/tests/web/api/test_custom_api_edit_lock_postgresql.py index 1677be232..750f65b3c 100644 --- a/tests/web/api/test_custom_api_edit_lock_postgresql.py +++ b/tests/web/api/test_custom_api_edit_lock_postgresql.py @@ -45,6 +45,7 @@ from xagent.web.models.database import Base from xagent.web.models.user import User from xagent.web.services.connector_team_scope import ( + ConnectorAccess, ConnectorDeleteDecision, set_connector_team_hooks, ) @@ -90,6 +91,51 @@ def seeded(session_factory): return int(owner.id), int(api.id) +@pytest.fixture() +def member(session_factory): + """A second user, distinct from ``seeded``'s owner, with no personal + link row of their own by default -- the caller the post-lock + re-authorization tests below exercise.""" + with session_factory() as db: + user = User( + username="custom-api-edit-lock-member", password_hash="x", is_admin=False + ) + db.add(user) + db.commit() + return int(user.id) + + +def _add_member_link(session_factory, member_id, api_id, *, can_edit): + with session_factory() as db: + db.add( + UserCustomApi( + user_id=member_id, + custom_api_id=api_id, + is_owner=False, + can_edit=can_edit, + is_active=True, + ) + ) + db.commit() + + +def _sequenced_access_hook(*answers): + """An access hook that answers differently on successive calls, mirroring + ``tests/web/api/test_custom_api_team_connector_edit.py``'s helper of the + same name. Records every call's ``refs`` on ``.calls`` so a test can pin + how many round trips the route pays.""" + calls: list[object] = [] + + def hook(db, user_id, refs): + calls.append(refs) + index = min(len(calls) - 1, len(answers) - 1) + answer = answers[index] + return {ref: answer for ref in refs} + + hook.calls = calls + return hook + + def test_a_second_editor_blocks_until_the_first_editors_transaction_finishes( session_factory, seeded ) -> None: @@ -933,3 +979,313 @@ def spy_deleted_hook(_db, _user_id, _connector_type, connector_id): assert fresh.query(CustomApi).filter(CustomApi.id == api_id).one(), ( "the shared definition row must survive a refused delete" ) + + +def test_a_link_deleted_mid_wait_is_admitted_when_the_team_verdict_still_grants( + session_factory, seeded, member +) -> None: + """A caller whose own link row grants the edit at the gate -- so the + gate never resolves a team verdict at all -- but whose row is deleted + by a second connection while this route holds the definition row's + lock. The post-lock re-check must ask the team verdict on its own + behalf, and a granting answer must let the edit through even though no + personal row survived the wait. + """ + import xagent.web.api.custom_api as custom_api_api + from xagent.web.api.custom_api import CustomApiUpdate + + owner_id, api_id = seeded + _add_member_link(session_factory, member, api_id, can_edit=True) + current_user = SimpleNamespace(id=member, is_admin=False) + + db = session_factory() + real_commit = db.commit + commits: list[str] = [] + + def record_commit(): + commits.append("commit") + return real_commit() + + db.commit = record_commit + revoked_already, queried_entities = _revoke_link_after_lock( + db, session_factory, member, api_id, "link-deleted", "can_edit" + ) + hook = _sequenced_access_hook(ConnectorAccess(team_owned=True, can_edit=True)) + set_connector_team_hooks(access=hook) + try: + response = custom_api_api.update_custom_api( + api_id, + CustomApiUpdate(description="admitted-after-link-deleted"), + current_user=current_user, + db=db, + ) + assert response.description == "admitted-after-link-deleted" + assert revoked_already.is_set(), "the concurrent deletion never ran" + assert queried_entities[:3] == [ + (UserCustomApi,), + (CustomApi,), + (UserCustomApi,), + ] + assert len(hook.calls) == 1, ( + "the gate's own row already granted the edit, so it never " + "resolves a verdict; only the post-lock re-check should ask" + ) + assert commits == ["commit"] + finally: + set_connector_team_hooks() + db.close() + + with session_factory() as fresh: + row = fresh.query(CustomApi).filter(CustomApi.id == api_id).one() + assert row.description == "admitted-after-link-deleted" + + +def test_a_link_deleted_mid_wait_after_a_granting_gate_verdict_costs_two_calls( + session_factory, seeded, member +) -> None: + """The gate itself needed the team verdict here (the caller's own row + does not grant the edit on its own), so it already spent one hook call + before the lock. The link is then deleted while the lock is held, and + the post-lock re-check spends a second call re-asking the same + verdict -- two calls in total, not one. + """ + import xagent.web.api.custom_api as custom_api_api + from xagent.web.api.custom_api import CustomApiUpdate + + owner_id, api_id = seeded + _add_member_link(session_factory, member, api_id, can_edit=False) + current_user = SimpleNamespace(id=member, is_admin=False) + + db = session_factory() + revoked_already, queried_entities = _revoke_link_after_lock( + db, session_factory, member, api_id, "link-deleted", "can_edit" + ) + hook = _sequenced_access_hook(ConnectorAccess(team_owned=True, can_edit=True)) + set_connector_team_hooks(access=hook) + try: + response = custom_api_api.update_custom_api( + api_id, + CustomApiUpdate(description="admitted-after-second-verdict"), + current_user=current_user, + db=db, + ) + assert response.description == "admitted-after-second-verdict" + assert revoked_already.is_set(), "the concurrent deletion never ran" + assert len(hook.calls) == 2, ( + "the gate's own can_edit=False row already spent one call; the " + "post-lock re-check must spend a second one, not reuse the first" + ) + finally: + set_connector_team_hooks() + db.close() + + with session_factory() as fresh: + row = fresh.query(CustomApi).filter(CustomApi.id == api_id).one() + assert row.description == "admitted-after-second-verdict" + + +def test_a_personal_row_downgraded_mid_wait_is_admitted_when_the_team_verdict_still_grants( + session_factory, seeded, member +) -> None: + """The caller's own row survives the wait but is downgraded to + ``can_edit=False`` by a second connection while the lock is held. The + gate never resolved a team verdict (the row granted the edit on its + own at that point), so this is the re-check's only call. + """ + import xagent.web.api.custom_api as custom_api_api + from xagent.web.api.custom_api import CustomApiUpdate + + owner_id, api_id = seeded + _add_member_link(session_factory, member, api_id, can_edit=True) + current_user = SimpleNamespace(id=member, is_admin=False) + + db = session_factory() + revoked_already, queried_entities = _revoke_link_after_lock( + db, session_factory, member, api_id, "can-edit-cleared", "can_edit" + ) + hook = _sequenced_access_hook(ConnectorAccess(team_owned=True, can_edit=True)) + set_connector_team_hooks(access=hook) + try: + response = custom_api_api.update_custom_api( + api_id, + CustomApiUpdate(description="admitted-after-downgrade"), + current_user=current_user, + db=db, + ) + assert response.description == "admitted-after-downgrade" + assert revoked_already.is_set(), "the concurrent downgrade never ran" + assert len(hook.calls) == 1 + finally: + set_connector_team_hooks() + db.close() + + with session_factory() as fresh: + row = fresh.query(CustomApi).filter(CustomApi.id == api_id).one() + assert row.description == "admitted-after-downgrade" + link = ( + fresh.query(UserCustomApi) + .filter( + UserCustomApi.custom_api_id == api_id, + UserCustomApi.user_id == member, + ) + .one() + ) + assert link.can_edit is False + + +def test_a_link_deleted_mid_wait_with_a_denying_verdict_is_a_404_with_no_shared_write( + session_factory, seeded, member +) -> None: + """The same deletion-mid-wait window as the granting tests above, but + the re-resolved team verdict denies the edit: the caller has no + surviving personal row and no team access either, so this is the same + 404 an unrelated caller has always gotten, with nothing committed. + """ + import xagent.web.api.custom_api as custom_api_api + from xagent.web.api.custom_api import CustomApiUpdate + + owner_id, api_id = seeded + _add_member_link(session_factory, member, api_id, can_edit=True) + current_user = SimpleNamespace(id=member, is_admin=False) + + db = session_factory() + real_commit = db.commit + commits: list[str] = [] + + def record_commit(): + commits.append("commit") + return real_commit() + + db.commit = record_commit + revoked_already, queried_entities = _revoke_link_after_lock( + db, session_factory, member, api_id, "link-deleted", "can_edit" + ) + hook = _sequenced_access_hook(ConnectorAccess(team_owned=True, can_edit=False)) + set_connector_team_hooks(access=hook) + try: + with pytest.raises(HTTPException) as exc: + custom_api_api.update_custom_api( + api_id, + CustomApiUpdate(description="should-not-land"), + current_user=current_user, + db=db, + ) + assert exc.value.status_code == 404 + assert revoked_already.is_set(), "the concurrent deletion never ran" + assert commits == [], "a refused edit must commit nothing" + finally: + set_connector_team_hooks() + db.close() + + with session_factory() as fresh: + row = fresh.query(CustomApi).filter(CustomApi.id == api_id).one() + assert row.description is None, ( + "the shared definition row must be untouched by a refused edit" + ) + + +def test_a_link_deleted_mid_wait_on_a_mixed_payload_is_400_not_a_stale_data_error( + session_factory, seeded, member +) -> None: + """A payload that carries ``is_active`` alongside a definition-row + field takes the lock (the definition field decides that), and the + caller's link row is then deleted while the lock is held. The team + verdict still grants the edit, so the definition-row half is allowed + through -- but with no personal row left to hold ``is_active``, the + locked guard must refuse with the same 400 the pre-lock guard uses, + rather than letting ``user_api.is_active = ...`` set a shadow + attribute on an ORM object backed by a row that no longer exists. + """ + import xagent.web.api.custom_api as custom_api_api + from xagent.web.api.custom_api import CustomApiUpdate + + owner_id, api_id = seeded + _add_member_link(session_factory, member, api_id, can_edit=True) + current_user = SimpleNamespace(id=member, is_admin=False) + + db = session_factory() + real_commit = db.commit + commits: list[str] = [] + + def record_commit(): + commits.append("commit") + return real_commit() + + db.commit = record_commit + revoked_already, queried_entities = _revoke_link_after_lock( + db, session_factory, member, api_id, "link-deleted", "can_edit" + ) + hook = _sequenced_access_hook(ConnectorAccess(team_owned=True, can_edit=True)) + set_connector_team_hooks(access=hook) + try: + with pytest.raises(HTTPException) as exc: + custom_api_api.update_custom_api( + api_id, + CustomApiUpdate(description="should-not-land", is_active=False), + current_user=current_user, + db=db, + ) + assert exc.value.status_code == 400 + assert exc.value.detail == ( + "No personal connection exists to configure is_active for this API" + ) + assert revoked_already.is_set(), "the concurrent deletion never ran" + assert commits == [], "a refused edit must commit nothing" + finally: + set_connector_team_hooks() + db.close() + + with session_factory() as fresh: + row = fresh.query(CustomApi).filter(CustomApi.id == api_id).one() + assert row.description is None, ( + "the shared definition row must be untouched by a refused edit" + ) + + +def test_a_link_deleted_mid_wait_admits_through_a_fresh_stand_in_not_a_stale_object( + session_factory, seeded, member +) -> None: + """The same admission as the first test in this group, checked from + the response side: with no personal row surviving the wait, the + response must come from a freshly constructed stand-in rather than + the (now stale) ORM object the gate resolved before the lock. Reusing + that stale object would raise ``ObjectDeletedError`` once this + request's own ``db.commit()`` expires it and something then reads one + of its columns -- the failure mode this route had before the fix, + which only shows up once this session's own commit forces a refresh + of an object mapped to a row a *different* connection removed. + """ + import xagent.web.api.custom_api as custom_api_api + from xagent.web.api.custom_api import CustomApiUpdate + + owner_id, api_id = seeded + _add_member_link(session_factory, member, api_id, can_edit=True) + current_user = SimpleNamespace(id=member, is_admin=False) + + db = session_factory() + revoked_already, queried_entities = _revoke_link_after_lock( + db, session_factory, member, api_id, "link-deleted", "can_edit" + ) + hook = _sequenced_access_hook(ConnectorAccess(team_owned=True, can_edit=True)) + set_connector_team_hooks(access=hook) + try: + response = custom_api_api.update_custom_api( + api_id, + CustomApiUpdate(description="stand-in-response"), + current_user=current_user, + db=db, + ) + assert revoked_already.is_set(), "the concurrent deletion never ran" + # The stand-in's own flag defaults, not whatever the deleted row + # last held -- read straight off the response with no further + # session activity in between, so a reused stale object would + # already have raised by this point rather than merely disagreeing. + assert response.is_active is True + assert response.is_default is False + finally: + set_connector_team_hooks() + db.close() + + with session_factory() as fresh: + row = fresh.query(CustomApi).filter(CustomApi.id == api_id).one() + assert row.description == "stand-in-response" diff --git a/tests/web/api/test_custom_api_team_connector_edit.py b/tests/web/api/test_custom_api_team_connector_edit.py index 5fc798e60..fd675f7e2 100644 --- a/tests/web/api/test_custom_api_team_connector_edit.py +++ b/tests/web/api/test_custom_api_team_connector_edit.py @@ -31,6 +31,7 @@ from xagent.web.models.user import User from xagent.web.services.connector_team_scope import ( ConnectorAccess, + ConnectorHookSessionBoundaryError, set_connector_team_hooks, snapshot_connector_team_hooks, ) @@ -705,3 +706,161 @@ def test_an_owner_pays_zero_calls(self, db): _put(api_id, CustomApiUpdate(description="owner-edit"), owner, db) assert len(hook.calls) == 0 + + +class TestPostLockRecheckDeclaresTheLock: + def test_a_hook_that_commits_on_the_post_lock_call_is_refused_as_a_boundary_violation( + self, db + ): + """The post-lock re-check declares ``caller_holds_lock=True`` on the + call it makes through ``_recheck_team_access_under_definition_lock``: a hook + that ends this request's own transaction while it holds the + definition row's lock must be refused as the seam's own boundary + violation, not answered as if the lock were still intact. + """ + owner = _make_user(db, 1) + member = _make_user(db, 2) + api = _make_owned_api(db, owner.id, name="boundary-probe") + api_id = api.id + original_description = ( + str(api.description) if api.description is not None else None + ) + + calls: list[object] = [] + + def hook(db, user_id, refs): + calls.append(refs) + if len(calls) == 2: + db.commit() + return { + ref: ConnectorAccess(team_owned=True, can_edit=True) for ref in refs + } + + with snapshot_connector_team_hooks(): + set_connector_team_hooks(access=hook) + with pytest.raises(ConnectorHookSessionBoundaryError): + _put( + api_id, + CustomApiUpdate(description="should-not-land"), + member, + db, + ) + + assert len(calls) == 2 + db.rollback() + refreshed = db.query(CustomApi).filter(CustomApi.id == api_id).one() + assert refreshed.description == original_description + + +class TestHookCallCountAcrossPayloadShapes: + """How many times a single request calls the access hook, across the + two shapes not already covered by ``TestTheRecheckCostsExactlyOneExtraHookCall`` + above: a payload that writes only ``is_active`` never takes the lock, so + a personal row that needs the team verdict to grant the edit asks once, + at the gate, and never again; and a deployment with no access hook + installed asks nothing at either point, because the hook slot being + empty answers ``{}`` without ever reaching the ``hook`` object a test + installs. The other two shapes -- an owner's row that already grants + the edit, and a stand-in caller who writes the shared definition row -- + are the same call counts ``TestTheRecheckCostsExactlyOneExtraHookCall`` + already covers above. + """ + + def test_is_active_only_payload_from_a_can_edit_false_personal_row_pays_one_call( + self, db + ): + owner = _make_user(db, 1) + member = _make_user(db, 2) + api = _make_owned_api(db, owner.id, name="count-is-active-only") + api_id = api.id + db.add( + UserCustomApi( + user_id=member.id, + custom_api_id=api_id, + is_owner=False, + can_edit=False, + is_active=True, + ) + ) + db.commit() + hook = _sequenced_access_hook(ConnectorAccess(team_owned=True, can_edit=True)) + + with snapshot_connector_team_hooks(): + set_connector_team_hooks(access=hook) + _put(api_id, CustomApiUpdate(is_active=False), member, db) + + assert len(hook.calls) == 1 + + def test_no_hook_installed_pays_zero_calls_regardless_of_payload(self, db): + owner = _make_user(db, 1) + member = _make_user(db, 2) + api = _make_owned_api(db, owner.id, name="count-no-hook") + api_id = api.id + # Never installed through ``set_connector_team_hooks``, so nothing + # in the seam can reach it; used only to prove that fact rather + # than to answer anything. + hook = _sequenced_access_hook(ConnectorAccess(team_owned=True, can_edit=True)) + + with snapshot_connector_team_hooks(): + set_connector_team_hooks() + with pytest.raises(HTTPException) as exc: + _put( + api_id, + CustomApiUpdate(description="stand-in-write"), + member, + db, + ) + + # No access hook installed means the caller's team can never be + # shown to link this connector, so the pre-lock resolution itself + # refuses with the same 404 an unrelated caller with no personal + # row has always gotten, before this route's own 403 is reached. + assert exc.value.status_code == 404 + assert len(hook.calls) == 0 + + +class TestReachablePathAWritesIsActiveWithoutRetakingTheLock: + def test_a_can_edit_false_personal_row_widened_by_the_team_writes_is_active_in_one_call( + self, db + ): + """The reachable reading of the gap this route used to have: a + caller with a personal row whose own ``can_edit`` is ``False``, who + can still write ``is_active`` because the team verdict grants the + edit. The payload sets no field of the shared definition row, so it + never takes that row's lock and never runs the post-lock re-check + -- there is nothing to re-establish, because nothing that could + move under an unbounded wait was ever locked. + """ + owner = _make_user(db, 1) + member = _make_user(db, 2) + api = _make_owned_api(db, owner.id, name="reachable-path-a") + api_id = api.id + db.add( + UserCustomApi( + user_id=member.id, + custom_api_id=api_id, + is_owner=False, + can_edit=False, + is_active=True, + ) + ) + db.commit() + hook = _sequenced_access_hook(ConnectorAccess(team_owned=True, can_edit=True)) + + with snapshot_connector_team_hooks(): + set_connector_team_hooks(access=hook) + response = _put(api_id, CustomApiUpdate(is_active=False), member, db) + + assert response.is_active is False + assert len(hook.calls) == 1 + + db.rollback() + link = ( + db.query(UserCustomApi) + .filter( + UserCustomApi.user_id == member.id, + UserCustomApi.custom_api_id == api_id, + ) + .one() + ) + assert link.is_active is False diff --git a/tests/web/services/test_connector_team_scope.py b/tests/web/services/test_connector_team_scope.py index fff733b04..f75e95411 100644 --- a/tests/web/services/test_connector_team_scope.py +++ b/tests/web/services/test_connector_team_scope.py @@ -2045,9 +2045,11 @@ async def test_the_boundary_handler_answers_500_and_one_detail(): def test_the_access_slot_lets_the_boundary_error_through_its_wrapper(db_session): - """``access`` has no call site in this repository today, so this is - constructed directly against the wrapper rather than through a route. - The seam's own transient-outage error gets folded into + """Constructed directly against the wrapper rather than through a route, + because the only ``access`` call site that holds a lock lives in + ``custom_api.py``, and that path is covered by + ``tests/web/api/test_custom_api_team_connector_edit.py`` instead. The + seam's own transient-outage error gets folded into ``ConnectorRuntimeError`` by the surrounding ``except Exception``; this one must not -- a permanent defect in the installing application's code is a different failure than an outage, and folding it in would From 06e44d583f5f4926e250b698a539ac319926ceeb Mon Sep 17 00:00:00 2001 From: Alexliu Date: Mon, 7 Sep 2026 01:20:39 +0800 Subject: [PATCH 08/10] fix(connectors): refuse is_active by presence and seal the stand-in objects Both is_active guards on the Custom API edit route checked whether the value was not None, but the surrounding contract is that any request carrying the field must be refused for a caller with no personal row. An explicit {"is_active": null} carries the field while being None, so it slipped past both guards silently instead of getting the refusal every other value already got. Both guards now check field presence via model_fields_set, the same test writes_definition_row already uses a few lines below. _TeamOwnedUserMCP and _TeamOwnedUserApi stand in for a caller with no personal association row and answer every response field from class attributes. Neither backs a database row, so a write reaching one of them would only create a shadowing instance attribute that persists nothing while a later read reports it back as real. Both classes now declare __slots__ = ("user_id",), so a write to any other attribute raises AttributeError instead of succeeding silently. --- src/xagent/web/api/custom_api.py | 22 ++++-- src/xagent/web/api/mcp.py | 23 +++++- .../test_custom_api_team_connector_edit.py | 73 ++++++++++++++++++- 3 files changed, 107 insertions(+), 11 deletions(-) diff --git a/src/xagent/web/api/custom_api.py b/src/xagent/web/api/custom_api.py index 04c589cbc..a5e81690a 100644 --- a/src/xagent/web/api/custom_api.py +++ b/src/xagent/web/api/custom_api.py @@ -490,12 +490,16 @@ def update_custom_api( ) # is_active lives on the personal association row; a caller with no - # personal row (the stand-in) has none to hold it, so a payload carrying - # it must be rejected outright -- writing it onto the stand-in would only - # set a shadowing instance attribute that persists nothing, and the - # response below would then read that shadow back and report a change that - # never happened. - if is_stand_in and api_data.is_active is not None: + # personal row (the stand-in) has none to hold it, so a request that + # carries the field at all must be rejected outright -- writing it onto + # the stand-in would only set a shadowing instance attribute that + # persists nothing, and the response below would then read that shadow + # back and report a change that never happened. This checks whether the + # field is present, the same test ``writes_definition_row`` below makes + # with ``model_fields_set``, not whether its value is ``None``: an + # explicit ``{"is_active": null}`` carries the field and must be refused + # the same as any other value would be. + if is_stand_in and "is_active" in api_data.model_fields_set: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="No personal connection exists to configure is_active for this API", @@ -713,8 +717,10 @@ def update_custom_api( # (see the ``is_stand_in`` check earlier in this route) catches a # caller who already had no personal row when the gate ran; this one # catches a caller whose personal row existed then and was deleted - # while this request waited for the lock. - if current_user_api is None and api_data.is_active is not None: + # while this request waited for the lock. Same presence test as the + # guard above: the field being in the request is what matters, not + # its value. + if current_user_api is None and "is_active" in api_data.model_fields_set: db.rollback() raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, diff --git a/src/xagent/web/api/mcp.py b/src/xagent/web/api/mcp.py index 49b7f4035..16cb0a57e 100644 --- a/src/xagent/web/api/mcp.py +++ b/src/xagent/web/api/mcp.py @@ -2001,7 +2001,18 @@ def _global_config_tampered(server_data: MCPServerUpdate, server: MCPServer) -> class _TeamOwnedUserMCP: """Stand-in for a missing UserMCPServer row: a team connector the user does not personally own. Exposes the attributes the response builders read with - not-owned defaults (usable, but not editable/deletable).""" + not-owned defaults (usable, but not editable/deletable). + + ``__slots__`` declares only ``user_id`` as a real per-instance attribute. + Every other name below is a class attribute, not a slot, so assigning to + it on an instance -- ``stand_in.is_active = False``, say -- raises + ``AttributeError`` instead of silently creating a shadowing instance + attribute the caller's own row never backs. A caller admitted through + this stand-in has no association row to write, so nothing here should + ever be writable. + """ + + __slots__ = ("user_id",) is_owner = False can_edit = False @@ -2016,7 +2027,15 @@ def __init__(self, user_id: int) -> None: class _TeamOwnedUserApi: - """Stand-in for a missing UserCustomApi row (team-owned, not user-owned).""" + """Stand-in for a missing UserCustomApi row (team-owned, not user-owned). + + Same reasoning as ``_TeamOwnedUserMCP`` above: ``__slots__`` leaves + ``user_id`` as the only attribute an instance can hold, so a write to + ``can_edit``, ``is_active`` or ``is_default`` raises ``AttributeError`` + rather than shadowing the class default with a value nothing persists. + """ + + __slots__ = ("user_id",) can_edit = False is_active = True diff --git a/tests/web/api/test_custom_api_team_connector_edit.py b/tests/web/api/test_custom_api_team_connector_edit.py index fd675f7e2..9de728a61 100644 --- a/tests/web/api/test_custom_api_team_connector_edit.py +++ b/tests/web/api/test_custom_api_team_connector_edit.py @@ -25,7 +25,11 @@ get_custom_api, update_custom_api, ) -from xagent.web.api.mcp import _custom_api_to_mcp_response, _TeamOwnedUserApi +from xagent.web.api.mcp import ( + _custom_api_to_mcp_response, + _TeamOwnedUserApi, + _TeamOwnedUserMCP, +) from xagent.web.models.custom_api import CustomApi, UserCustomApi from xagent.web.models.database import Base from xagent.web.models.user import User @@ -385,6 +389,35 @@ def test_is_active_from_a_caller_with_no_personal_row_is_400_not_a_silent_drop( refreshed = db.query(CustomApi).filter(CustomApi.id == api_id).one() assert refreshed.name == "unchanged-name" + def test_explicit_null_is_active_from_a_stand_in_caller_is_also_400(self, db): + """The guard tests whether ``is_active`` is present in the request, + not whether its value is ``None`` -- the same distinction + ``writes_definition_row`` draws with ``model_fields_set`` elsewhere + in this route. ``CustomApiUpdate(is_active=None)`` carries the field + explicitly, so it must be refused exactly like + ``CustomApiUpdate(is_active=False)`` above, not treated as if the + field were simply absent from the payload. + """ + owner = _make_user(db, 1) + editor = _make_user(db, 2) + api = _make_owned_api(db, owner.id, name="unchanged-name-null-payload") + api_id = api.id + + payload = CustomApiUpdate(is_active=None) + assert "is_active" in payload.model_fields_set + + with snapshot_connector_team_hooks(): + set_connector_team_hooks( + access=lambda db, user_id, refs: { + ref: ConnectorAccess(team_owned=True, can_edit=True) for ref in refs + } + ) + with pytest.raises(HTTPException) as exc: + _put(api_id, payload, editor, db) + + assert exc.value.status_code == 400 + assert "personal connection" in str(exc.value.detail) + class TestStandInFlagsAgreeAcrossBothResponseSurfaces: def test_a_stand_in_caller_gets_the_same_two_flags_from_both_surfaces(self, db): @@ -425,6 +458,44 @@ def test_a_stand_in_caller_gets_the_same_two_flags_from_both_surfaces(self, db): assert detail.is_default == aggregate.is_default +class TestStandInObjectsRejectAttributeWrites: + """``_TeamOwnedUserMCP`` and ``_TeamOwnedUserApi`` stand in for a caller + with no personal association row. Neither backs a database row, so + nothing about them should ever be writable: a write that reached one of + them would only set a shadowing instance attribute that persists + nothing and that a later read on the same object would report back as + if it had. ``__slots__`` on both classes makes that a hard + ``AttributeError`` instead of a silent instance attribute, for every + name that is not ``user_id``. + """ + + def test_writing_is_active_or_can_edit_on_a_stand_in_mcp_raises(self): + stand_in = _TeamOwnedUserMCP(7) + + with pytest.raises(AttributeError): + stand_in.is_active = False + with pytest.raises(AttributeError): + stand_in.can_edit = True + + assert _TeamOwnedUserMCP.is_active is True + assert _TeamOwnedUserMCP.can_edit is False + assert stand_in.is_active is True + assert stand_in.can_edit is False + + def test_writing_is_active_or_can_edit_on_a_stand_in_api_raises(self): + stand_in = _TeamOwnedUserApi(7) + + with pytest.raises(AttributeError): + stand_in.is_active = False + with pytest.raises(AttributeError): + stand_in.can_edit = True + + assert _TeamOwnedUserApi.is_active is True + assert _TeamOwnedUserApi.can_edit is False + assert stand_in.is_active is True + assert stand_in.can_edit is False + + class TestTypedErrorArm: """A raising hook still surfaces its declared status for a caller with no working personal row -- the verdict is genuinely the gate for that From 974281383c9bdc4272fa217e156d28291655d5fb Mon Sep 17 00:00:00 2001 From: Alexliu Date: Mon, 7 Sep 2026 01:21:37 +0800 Subject: [PATCH 09/10] test(connectors): drop the hook counter that was never installed test_no_hook_installed_pays_zero_calls_regardless_of_payload built a counting hook object but never passed it to set_connector_team_hooks, so hook.calls could never contain anything and the assertion on it could never fail regardless of the route's behaviour. The test still has a real assertion in it -- the 404 a stand-in caller gets when no access hook is installed at all -- so it keeps that and drops only the unwired counter and the claim built on it. Renamed to describe what it actually checks. --- .../test_custom_api_team_connector_edit.py | 29 ++++++++----------- 1 file changed, 12 insertions(+), 17 deletions(-) diff --git a/tests/web/api/test_custom_api_team_connector_edit.py b/tests/web/api/test_custom_api_team_connector_edit.py index 9de728a61..0b9096b34 100644 --- a/tests/web/api/test_custom_api_team_connector_edit.py +++ b/tests/web/api/test_custom_api_team_connector_edit.py @@ -824,17 +824,17 @@ def hook(db, user_id, refs): class TestHookCallCountAcrossPayloadShapes: - """How many times a single request calls the access hook, across the - two shapes not already covered by ``TestTheRecheckCostsExactlyOneExtraHookCall`` - above: a payload that writes only ``is_active`` never takes the lock, so - a personal row that needs the team verdict to grant the edit asks once, - at the gate, and never again; and a deployment with no access hook - installed asks nothing at either point, because the hook slot being - empty answers ``{}`` without ever reaching the ``hook`` object a test - installs. The other two shapes -- an owner's row that already grants - the edit, and a stand-in caller who writes the shared definition row -- - are the same call counts ``TestTheRecheckCostsExactlyOneExtraHookCall`` - already covers above. + """How many times a single request calls the access hook, for a payload + that writes only ``is_active`` and never takes the lock: a personal row + that needs the team verdict to grant the edit asks once, at the gate, + and never again. A deployment with no access hook installed is covered + alongside it below for the 404 it gives a stand-in caller, not for a + call count: an empty hook slot resolves to ``{}`` on its own, with no + hook object installed to call and so nothing to count. The other two + shapes -- an owner's row that already grants the edit, and a stand-in + caller who writes the shared definition row -- are the same call + counts ``TestTheRecheckCostsExactlyOneExtraHookCall`` already covers + above. """ def test_is_active_only_payload_from_a_can_edit_false_personal_row_pays_one_call( @@ -862,15 +862,11 @@ def test_is_active_only_payload_from_a_can_edit_false_personal_row_pays_one_call assert len(hook.calls) == 1 - def test_no_hook_installed_pays_zero_calls_regardless_of_payload(self, db): + def test_no_hook_installed_answers_404_for_a_stand_in_caller(self, db): owner = _make_user(db, 1) member = _make_user(db, 2) api = _make_owned_api(db, owner.id, name="count-no-hook") api_id = api.id - # Never installed through ``set_connector_team_hooks``, so nothing - # in the seam can reach it; used only to prove that fact rather - # than to answer anything. - hook = _sequenced_access_hook(ConnectorAccess(team_owned=True, can_edit=True)) with snapshot_connector_team_hooks(): set_connector_team_hooks() @@ -887,7 +883,6 @@ def test_no_hook_installed_pays_zero_calls_regardless_of_payload(self, db): # refuses with the same 404 an unrelated caller with no personal # row has always gotten, before this route's own 403 is reached. assert exc.value.status_code == 404 - assert len(hook.calls) == 0 class TestReachablePathAWritesIsActiveWithoutRetakingTheLock: From 26b0928d9973734716e9199862e8fd31e98ab3fb Mon Sep 17 00:00:00 2001 From: Alexliu Date: Mon, 7 Sep 2026 02:11:42 +0800 Subject: [PATCH 10/10] test(connectors): pin the explicit-null is_active refusal under the lock The post-lock guard refuses a payload that names is_active at all, not one whose value is non-null. Parametrize the mid-wait deletion test over False and None so the presence check under the lock has a red case of its own. --- .../api/test_custom_api_edit_lock_postgresql.py | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/tests/web/api/test_custom_api_edit_lock_postgresql.py b/tests/web/api/test_custom_api_edit_lock_postgresql.py index 750f65b3c..c7ba76548 100644 --- a/tests/web/api/test_custom_api_edit_lock_postgresql.py +++ b/tests/web/api/test_custom_api_edit_lock_postgresql.py @@ -1184,8 +1184,9 @@ def record_commit(): ) +@pytest.mark.parametrize("is_active_value", [False, None]) def test_a_link_deleted_mid_wait_on_a_mixed_payload_is_400_not_a_stale_data_error( - session_factory, seeded, member + session_factory, seeded, member, is_active_value ) -> None: """A payload that carries ``is_active`` alongside a definition-row field takes the lock (the definition field decides that), and the @@ -1195,6 +1196,13 @@ def test_a_link_deleted_mid_wait_on_a_mixed_payload_is_400_not_a_stale_data_erro locked guard must refuse with the same 400 the pre-lock guard uses, rather than letting ``user_api.is_active = ...`` set a shadow attribute on an ORM object backed by a row that no longer exists. + + Parametrized over the value ``is_active`` carries, because the guard + tests presence, not value: an explicit ``{"is_active": null}`` carries + the field just as ``false`` does and must be refused the same way. A + value-based guard (``api_data.is_active is not None``) lets the null + case through and commits the definition-row half for a caller with no + link row left to hold ``is_active``. """ import xagent.web.api.custom_api as custom_api_api from xagent.web.api.custom_api import CustomApiUpdate @@ -1217,11 +1225,13 @@ def record_commit(): ) hook = _sequenced_access_hook(ConnectorAccess(team_owned=True, can_edit=True)) set_connector_team_hooks(access=hook) + payload = CustomApiUpdate(description="should-not-land", is_active=is_active_value) + assert "is_active" in payload.model_fields_set try: with pytest.raises(HTTPException) as exc: custom_api_api.update_custom_api( api_id, - CustomApiUpdate(description="should-not-land", is_active=False), + payload, current_user=current_user, db=db, )