diff --git a/docsgpt/api/user/agents/routes.py b/docsgpt/api/user/agents/routes.py index c65b0df6..2ddcda5d 100644 --- a/docsgpt/api/user/agents/routes.py +++ b/docsgpt/api/user/agents/routes.py @@ -1456,6 +1456,33 @@ class DeleteAgent(Resource): return make_response(jsonify({"id": pg_agent_id}), 200) +def _user_may_pin(conn, agent: dict, user_id: str, shared_with_me: set) -> bool: + """Whether ``user_id`` may pin, and keep reading, ``agent``. + + Pinning follows visibility: the owner, a team grantee, anyone who opened a + still-live share link, and the premade system templates. Without this gate a + pin is an unscoped read of any agent id. + + Args: + conn: Open database connection. + agent: The agent row, needing at least ``id``, ``user_id`` and ``shared``. + user_id: The caller. + shared_with_me: Agent ids the caller reached through a share link. + + Returns: + True if the caller is allowed to see the agent. + """ + owner_id = agent.get("user_id") + if owner_id == user_id: + return True + if owner_id in ("system", "__system__"): + return True + agent_id = str(agent["id"]) + if agent.get("shared") and agent_id in shared_with_me: + return True + return can_access(conn, "agent", agent_id, user_id) + + @agents_ns.route("/pinned_agents") class PinnedAgents(Resource): @api.doc(description="Get pinned agents for the user") @@ -1469,11 +1496,13 @@ class PinnedAgents(Resource): with db_session() as conn: users_repo = UsersRepository(conn) user_doc = users_repo.upsert(user_id) - pinned_ids = ( - user_doc.get("agent_preferences", {}).get("pinned", []) + prefs = ( + user_doc.get("agent_preferences", {}) if isinstance(user_doc.get("agent_preferences"), dict) - else [] + else {} ) + pinned_ids = prefs.get("pinned", []) or [] + shared_with_me = set(prefs.get("shared_with_me", []) or []) if not pinned_ids: return make_response(jsonify([]), 200) @@ -1500,6 +1529,15 @@ class PinnedAgents(Resource): if stale: users_repo.remove_pinned_bulk(user_id, stale) + # A pin is not proof of access: a share or team grant can be + # revoked long after the agent was pinned, so re-check every + # row on read rather than trusting the stored id list. + pinned_agents = [ + agent + for agent in pinned_agents + if _user_may_pin(conn, agent, user_id, shared_with_me) + ] + list_pinned_agents = [] for agent in pinned_agents: source_id = agent.get("source_id") @@ -1528,7 +1566,7 @@ class PinnedAgents(Resource): "last_used_at": agent.get("last_used_at", ""), "key": ( f"{agent['key'][:4]}...{agent['key'][-4:]}" - if agent.get("key") + if agent.get("key") and agent.get("user_id") == user_id else "" ), "pinned": True, @@ -1797,20 +1835,25 @@ class PinAgent(Resource): ) try: with db_session() as conn: - # Any user can pin any agent they can see — including - # shared ones. Use the non-user-scoped lookup so pins - # aren't restricted to owner-only. + # Any user can pin any agent they can *see* — owned, team-shared, + # reached through a share link, or a system template. The lookup + # is deliberately not owner-scoped, so visibility is checked + # separately below. from sqlalchemy import text as _sql_text if looks_like_uuid(agent_id): agent_row = conn.execute( - _sql_text("SELECT id FROM agents WHERE id = CAST(:id AS uuid)"), + _sql_text( + "SELECT id, user_id, shared FROM agents " + "WHERE id = CAST(:id AS uuid)" + ), {"id": agent_id}, ).fetchone() else: agent_row = conn.execute( _sql_text( - "SELECT id FROM agents WHERE legacy_mongo_id = :id" + "SELECT id, user_id, shared FROM agents " + "WHERE legacy_mongo_id = :id" ), {"id": agent_id}, ).fetchone() @@ -1823,15 +1866,28 @@ class PinAgent(Resource): users_repo = UsersRepository(conn) user_doc = users_repo.upsert(user_id) - pinned_list = ( - user_doc.get("agent_preferences", {}).get("pinned", []) + prefs = ( + user_doc.get("agent_preferences", {}) if isinstance(user_doc.get("agent_preferences"), dict) - else [] + else {} ) + pinned_list = prefs.get("pinned", []) or [] if pg_agent_id in pinned_list: + # Unpinning stays open regardless of current access, or a + # revoked share would leave a pin the user can't clear. users_repo.remove_pinned(user_id, pg_agent_id) action = "unpinned" else: + if not _user_may_pin( + conn, + dict(agent_row._mapping), + user_id, + set(prefs.get("shared_with_me", []) or []), + ): + return make_response( + jsonify({"success": False, "message": "Agent not found"}), + 404, + ) users_repo.add_pinned(user_id, pg_agent_id) action = "pinned" except Exception as err: diff --git a/tests/api/user/agents/test_routes_happy.py b/tests/api/user/agents/test_routes_happy.py index 66e6c95f..d850970f 100644 --- a/tests/api/user/agents/test_routes_happy.py +++ b/tests/api/user/agents/test_routes_happy.py @@ -1503,6 +1503,138 @@ class TestPinAgentMore: assert response.status_code == 500 +class TestPinAccessControl: + """Pinning must follow visibility — it is a read of the agent.""" + + def _pin(self, app, pg_conn, agent_id, user): + from docsgpt.api.user.agents.routes import PinAgent + + with _patch_db(pg_conn), app.test_request_context( + f"/api/pin_agent?id={agent_id}", method="POST" + ): + from flask import request + request.decoded_token = {"sub": user} + return PinAgent().post() + + def _pinned(self, app, pg_conn, user): + from docsgpt.api.user.agents.routes import PinnedAgents + + with _patch_db(pg_conn), app.test_request_context("/api/pinned_agents"): + from flask import request + request.decoded_token = {"sub": user} + return PinnedAgents().get() + + def test_stranger_cannot_pin_another_users_agent(self, app, pg_conn): + from docsgpt.storage.db.repositories.users import UsersRepository + + owner, stranger = "u-pin-owner", "u-pin-stranger" + agent = _seed_agent(pg_conn, user=owner, name="Private") + + response = self._pin(app, pg_conn, str(agent["id"]), stranger) + + assert response.status_code == 404 + prefs = UsersRepository(pg_conn).upsert(stranger).get( + "agent_preferences" + ) or {} + assert prefs.get("pinned", []) == [] + + def test_stranger_pinned_list_stays_empty(self, app, pg_conn): + """Even a pin forced into the prefs must not leak the agent.""" + from docsgpt.storage.db.repositories.users import UsersRepository + + owner, stranger = "u-pin-owner2", "u-pin-stranger2" + agent = _seed_agent(pg_conn, user=owner, name="Private") + users_repo = UsersRepository(pg_conn) + users_repo.upsert(stranger) + users_repo.add_pinned(stranger, str(agent["id"])) + + response = self._pinned(app, pg_conn, stranger) + + assert response.status_code == 200 + assert response.json == [] + + def test_share_link_recipient_can_pin(self, app, pg_conn): + from docsgpt.storage.db.repositories.agents import AgentsRepository + from docsgpt.storage.db.repositories.users import UsersRepository + + owner, recipient = "u-pin-sharer", "u-pin-recipient" + agent = _seed_agent(pg_conn, user=owner, name="Shared") + agent_id = str(agent["id"]) + AgentsRepository(pg_conn).update(agent_id, owner, {"shared": True}) + users_repo = UsersRepository(pg_conn) + users_repo.upsert(recipient) + users_repo.add_shared(recipient, agent_id) + + assert self._pin(app, pg_conn, agent_id, recipient).status_code == 200 + listed = self._pinned(app, pg_conn, recipient) + assert [a["id"] for a in listed.json] == [agent_id] + + def test_revoked_share_drops_out_of_the_pinned_list(self, app, pg_conn): + from docsgpt.storage.db.repositories.agents import AgentsRepository + from docsgpt.storage.db.repositories.users import UsersRepository + + owner, recipient = "u-pin-revoke-owner", "u-pin-revoke-recipient" + agent = _seed_agent(pg_conn, user=owner, name="Shared") + agent_id = str(agent["id"]) + repo = AgentsRepository(pg_conn) + repo.update(agent_id, owner, {"shared": True}) + users_repo = UsersRepository(pg_conn) + users_repo.upsert(recipient) + users_repo.add_shared(recipient, agent_id) + self._pin(app, pg_conn, agent_id, recipient) + + repo.update(agent_id, owner, {"shared": False}) + + assert self._pinned(app, pg_conn, recipient).json == [] + + def test_team_grantee_can_pin_but_gets_no_key(self, app, pg_conn): + from docsgpt.storage.db.repositories.agents import AgentsRepository + + owner, member = "u-pin-team-owner", "u-pin-team-member" + agent = _seed_agent(pg_conn, user=owner, name="Team") + agent_id = str(agent["id"]) + AgentsRepository(pg_conn).update(agent_id, owner, {"key": "abcd1234wxyz"}) + + with patch( + "docsgpt.api.user.agents.routes.can_access", return_value=True + ): + assert self._pin(app, pg_conn, agent_id, member).status_code == 200 + listed = self._pinned(app, pg_conn, member) + + assert [a["id"] for a in listed.json] == [agent_id] + assert listed.json[0]["key"] == "" + + def test_owner_still_sees_the_masked_key(self, app, pg_conn): + from docsgpt.storage.db.repositories.agents import AgentsRepository + + owner = "u-pin-key-owner" + agent = _seed_agent(pg_conn, user=owner, name="Mine") + agent_id = str(agent["id"]) + AgentsRepository(pg_conn).update(agent_id, owner, {"key": "abcd1234wxyz"}) + + self._pin(app, pg_conn, agent_id, owner) + + assert self._pinned(app, pg_conn, owner).json[0]["key"] == "abcd...wxyz" + + def test_unpin_survives_lost_access(self, app, pg_conn): + """A revoked share must not strand a pin the user can't remove.""" + from docsgpt.storage.db.repositories.users import UsersRepository + + owner, recipient = "u-unpin-owner", "u-unpin-recipient" + agent = _seed_agent(pg_conn, user=owner, name="Gone") + agent_id = str(agent["id"]) + users_repo = UsersRepository(pg_conn) + users_repo.upsert(recipient) + users_repo.add_pinned(recipient, agent_id) + + response = self._pin(app, pg_conn, agent_id, recipient) + + assert response.status_code == 200 + assert response.json["action"] == "unpinned" + prefs = users_repo.upsert(recipient).get("agent_preferences") or {} + assert agent_id not in prefs.get("pinned", []) + + class TestRegenerateAgentKey: def _seed_published_with_key(self, pg_conn, user, key): from docsgpt.storage.db.repositories.agents import AgentsRepository