diff --git a/tests/test_user_shares.py b/tests/test_user_shares.py new file mode 100644 index 000000000..2ecae7a5a --- /dev/null +++ b/tests/test_user_shares.py @@ -0,0 +1,396 @@ +"""Tests for the user-to-user resource sharing routes. + +Covers: + POST /api/shares — create share (idempotent, unknown user, self-share) + POST /api/shares/{id}/accept — accept-gate (target-only, already-decided, wrong-user) + POST /api/shares/{id}/deny — deny-gate (target-only, already-decided) + DELETE /api/shares/{id} — revoke (owner, not-found) + GET /api/shares — list (out/in) + user_can_access — gates on status='accepted' +""" +from __future__ import annotations + +import pytest +import pytest_asyncio +from httpx import ASGITransport, AsyncClient + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest_asyncio.fixture +async def shares_client(client, tmp_data_dir): + """Async client with user_shares store initialised, authenticated as admin. + + Builds on the conftest client fixture (which handles auth + store init). + """ + from tinyagentos.user_shares_store import UserSharesStore + import secrets + + app = client._transport.app + + # Init user_shares store (lifespan not running in tests). + store = UserSharesStore(tmp_data_dir / "user_shares.db") + await store.init() + app.state.user_shares = store + + # Create a target user via invite flow. + auth = app.state.auth + admin_record = auth.find_user("admin") + admin_uid = admin_record["id"] if admin_record else "" + + target_record = auth.find_user("target") + if target_record is None: + invite_code = auth.add_user_invite("target", "admin") + auth.complete_invite("target", invite_code, "Target User", "", "targetpass") + target_record = auth.find_user("target") + target_uid = target_record["id"] if target_record else "" + + # Set CSRF token so POST/PUT/DELETE routes pass verify_csrf. + csrf_token = secrets.token_hex(32) + client.cookies["csrf_token"] = csrf_token + client.headers["X-CSRF-Token"] = csrf_token + + client._admin_uid = admin_uid + client._target_uid = target_uid + + yield client + + await store.close() + + +@pytest_asyncio.fixture +async def shares_client_target(shares_client): + """Async client authenticated as the target user. + + Reuses shares_client's setup and just swaps session to target user. + """ + from httpx import ASGITransport, AsyncClient + import secrets + + app = shares_client._transport.app + auth = app.state.auth + target_uid = shares_client._target_uid + target_token = auth.create_session(user_id=target_uid, long_lived=True) + + # Set CSRF token. + csrf_token = secrets.token_hex(32) + + transport = ASGITransport(app=app) + async with AsyncClient( + transport=transport, + base_url="http://test", + cookies={"taos_session": target_token, "csrf_token": csrf_token}, + headers={"X-CSRF-Token": csrf_token}, + ) as c: + c._target_uid = target_uid + c._admin_uid = shares_client._admin_uid + yield c + + +# --------------------------------------------------------------------------- +# Route tests +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +class TestShareRoutes: + + # -- Create ---------------------------------------------------------- + + async def test_create_share_returns_record(self, shares_client): + """POST /api/shares creates a share and returns the record with status='pending'.""" + resp = await shares_client.post( + "/api/shares", + json={ + "resource_type": "project", + "resource_id": "proj-1", + "to_username": "target", + "permission": "read", + }, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["owner_user_id"] == shares_client._admin_uid + assert data["shared_with_user_id"] == shares_client._target_uid + assert data["resource_type"] == "project" + assert data["resource_id"] == "proj-1" + assert data["permission"] == "read" + assert data.get("status") == "pending" + assert "id" in data + + async def test_create_share_idempotent(self, shares_client): + """Re-sharing the same resource+target+permission is idempotent (no duplicates).""" + body = { + "resource_type": "project", + "resource_id": "proj-2", + "to_username": "target", + "permission": "read", + } + r1 = await shares_client.post("/api/shares", json=body) + assert r1.status_code == 200 + + r2 = await shares_client.post("/api/shares", json=body) + assert r2.status_code == 200 + + # Verify only one share exists for this resource — no duplicates. + resp = await shares_client.get("/api/shares?direction=out") + assert resp.status_code == 200 + matching = [ + s for s in resp.json() + if s["resource_id"] == "proj-2" and s["resource_type"] == "project" + ] + assert len(matching) == 1 + + async def test_create_share_unknown_user_returns_404(self, shares_client): + """Sharing with a non-existent username returns 404.""" + resp = await shares_client.post( + "/api/shares", + json={ + "resource_type": "project", + "resource_id": "proj-3", + "to_username": "nosuchuser", + "permission": "read", + }, + ) + assert resp.status_code == 404 + assert "nosuchuser" in resp.json()["detail"] + + async def test_create_share_self_share_returns_400(self, shares_client): + """Sharing with yourself returns 400.""" + resp = await shares_client.post( + "/api/shares", + json={ + "resource_type": "project", + "resource_id": "proj-4", + "to_username": "admin", # same user + "permission": "read", + }, + ) + assert resp.status_code == 400 + assert "yourself" in resp.json()["detail"] + + async def test_create_share_invalid_resource_type_returns_422(self, shares_client): + """POST with an unsupported resource_type returns 422.""" + resp = await shares_client.post( + "/api/shares", + json={ + "resource_type": "bogus_type", + "resource_id": "proj-422-rt", + "to_username": "target", + "permission": "read", + }, + ) + assert resp.status_code == 422 + assert "invalid resource_type" in resp.json()["detail"] + + async def test_create_share_invalid_permission_returns_422(self, shares_client): + """POST with an unsupported permission returns 422.""" + resp = await shares_client.post( + "/api/shares", + json={ + "resource_type": "project", + "resource_id": "proj-422-perm", + "to_username": "target", + "permission": "bogus_perm", + }, + ) + assert resp.status_code == 422 + assert "invalid permission" in resp.json()["detail"] + + # -- Accept ---------------------------------------------------------- + + async def test_accept_share_target_user(self, shares_client, shares_client_target): + """Target user can accept a pending share; user_can_access then returns True.""" + # Create share as admin → target. + r = await shares_client.post( + "/api/shares", + json={ + "resource_type": "project", + "resource_id": "proj-accept", + "to_username": "target", + "permission": "read", + }, + ) + assert r.status_code == 200 + share_id = r.json()["id"] + assert r.json()["status"] == "pending" + + # Before accept, user_can_access returns False (status != 'accepted'). + store = shares_client._transport.app.state.user_shares + can = await store.user_can_access("project", "proj-accept", shares_client._target_uid) + assert can is False + + # Target user accepts the share. + resp = await shares_client_target.post(f"/api/shares/{share_id}/accept") + assert resp.status_code == 200 + assert resp.json()["status"] == "accepted" + + # After accept, user_can_access returns True. + can = await store.user_can_access("project", "proj-accept", shares_client._target_uid) + assert can is True + + async def test_accept_share_wrong_user(self, shares_client): + """Only the target user can accept a share — admin trying returns 403.""" + r = await shares_client.post( + "/api/shares", + json={ + "resource_type": "project", + "resource_id": "proj-accept-wrong", + "to_username": "target", + "permission": "read", + }, + ) + assert r.status_code == 200 + share_id = r.json()["id"] + + # Admin (not the target) tries to accept → 403. + resp = await shares_client.post(f"/api/shares/{share_id}/accept") + assert resp.status_code == 403 + assert "only the target user" in resp.json()["detail"] + + async def test_accept_share_already_decided(self, shares_client, shares_client_target): + """Accepting an already-accepted share returns 409.""" + r = await shares_client.post( + "/api/shares", + json={ + "resource_type": "project", + "resource_id": "proj-accept-twice", + "to_username": "target", + "permission": "read", + }, + ) + assert r.status_code == 200 + share_id = r.json()["id"] + + # First accept succeeds. + r1 = await shares_client_target.post(f"/api/shares/{share_id}/accept") + assert r1.status_code == 200 + + # Second accept returns 409. + r2 = await shares_client_target.post(f"/api/shares/{share_id}/accept") + assert r2.status_code == 409 + + # -- Deny ------------------------------------------------------------ + + async def test_deny_share_target_user(self, shares_client, shares_client_target): + """Target user can deny a pending share; status becomes 'denied'.""" + r = await shares_client.post( + "/api/shares", + json={ + "resource_type": "project", + "resource_id": "proj-deny", + "to_username": "target", + "permission": "read", + }, + ) + assert r.status_code == 200 + share_id = r.json()["id"] + + resp = await shares_client_target.post(f"/api/shares/{share_id}/deny") + assert resp.status_code == 200 + assert resp.json()["status"] == "denied" + + # After deny, user_can_access returns False. + store = shares_client._transport.app.state.user_shares + can = await store.user_can_access("project", "proj-deny", shares_client._target_uid) + assert can is False + + async def test_deny_share_already_decided(self, shares_client, shares_client_target): + """Denying an already-denied share returns 409.""" + r = await shares_client.post( + "/api/shares", + json={ + "resource_type": "project", + "resource_id": "proj-deny-twice", + "to_username": "target", + "permission": "read", + }, + ) + assert r.status_code == 200 + share_id = r.json()["id"] + + r1 = await shares_client_target.post(f"/api/shares/{share_id}/deny") + assert r1.status_code == 200 + + r2 = await shares_client_target.post(f"/api/shares/{share_id}/deny") + assert r2.status_code == 409 + + # -- Revoke ---------------------------------------------------------- + + async def test_revoke_share_owner(self, shares_client): + """Owner can revoke their own share; it is no longer listed.""" + r = await shares_client.post( + "/api/shares", + json={ + "resource_type": "project", + "resource_id": "proj-revoke", + "to_username": "target", + "permission": "read", + }, + ) + assert r.status_code == 200 + share_id = r.json()["id"] + + resp = await shares_client.delete(f"/api/shares/{share_id}") + assert resp.status_code == 200 + assert resp.json() == {"status": "revoked", "share_id": share_id} + + # Verify share no longer listed. + out = await shares_client.get("/api/shares?direction=out") + assert out.status_code == 200 + matching = [s for s in out.json() if s["id"] == share_id] + assert len(matching) == 0 + + async def test_revoke_share_not_found(self, shares_client): + """Revoking a non-existent share returns 404.""" + resp = await shares_client.delete("/api/shares/99999") + assert resp.status_code == 404 + + # -- List ------------------------------------------------------------ + + async def test_list_shares_out(self, shares_client): + """GET /api/shares?direction=out lists shares owned by the authenticated user.""" + # Create two shares. + for i, res_id in enumerate(["proj-list-out-1", "proj-list-out-2"]): + r = await shares_client.post( + "/api/shares", + json={ + "resource_type": "project", + "resource_id": res_id, + "to_username": "target", + "permission": "read", + }, + ) + assert r.status_code == 200 + + resp = await shares_client.get("/api/shares?direction=out") + assert resp.status_code == 200 + data = resp.json() + assert isinstance(data, list) + # At least the two we just created. + out_ids = [s["resource_id"] for s in data if s["resource_type"] == "project"] + assert "proj-list-out-1" in out_ids + assert "proj-list-out-2" in out_ids + + async def test_list_shares_in(self, shares_client, shares_client_target): + """GET /api/shares?direction=in lists shares received by the authenticated user.""" + # Create a share as admin → target. + r = await shares_client.post( + "/api/shares", + json={ + "resource_type": "project", + "resource_id": "proj-list-in", + "to_username": "target", + "permission": "read", + }, + ) + assert r.status_code == 200 + + # Target user lists incoming shares. + resp = await shares_client_target.get("/api/shares?direction=in") + assert resp.status_code == 200 + data = resp.json() + assert isinstance(data, list) + in_ids = [s["resource_id"] for s in data if s["resource_type"] == "project"] + assert "proj-list-in" in in_ids diff --git a/tinyagentos/app.py b/tinyagentos/app.py index f0ef43fe6..718dd42a5 100644 --- a/tinyagentos/app.py +++ b/tinyagentos/app.py @@ -425,6 +425,8 @@ async def _probe_backend(backend: dict) -> dict: shared_docs_store = SharedDocsStore(data_dir / "shared_docs.db") from tinyagentos.todo.todo_store import TodoStore todo_store = TodoStore(data_dir / "todo.db") + from tinyagentos.user_shares_store import UserSharesStore + user_shares_store = UserSharesStore(data_dir / "user_shares.db") from tinyagentos.coding_sessions.launcher import CodingSessionLauncher from tinyagentos.coding_sessions.store import CodingSessionStore coding_session_store = CodingSessionStore(data_dir / "coding_sessions.db") @@ -577,6 +579,9 @@ async def lifespan(app: FastAPI): await routine_store.init() await project_canvas_store.init() await decision_store.init() + app.state.decision_store = decision_store + await user_shares_store.init() + app.state.user_shares = user_shares_store await execution_policy_store.init() await shared_docs_store.init() await todo_store.init() diff --git a/tinyagentos/routes/__init__.py b/tinyagentos/routes/__init__.py index ac878879c..07e919ea0 100644 --- a/tinyagentos/routes/__init__.py +++ b/tinyagentos/routes/__init__.py @@ -7,7 +7,6 @@ def register_all_routers(app): """ from fastapi import Depends from tinyagentos.middleware.csrf import verify_csrf - _csrf = [Depends(verify_csrf)] from tinyagentos.routes.auth import router as auth_router @@ -413,3 +412,6 @@ def register_all_routers(app): from tinyagentos.routes.council import router as council_router app.include_router(council_router, dependencies=_csrf) + + from tinyagentos.routes.user_shares import router as user_shares_router + app.include_router(user_shares_router, dependencies=_csrf) diff --git a/tinyagentos/routes/user_shares.py b/tinyagentos/routes/user_shares.py new file mode 100644 index 000000000..a693a60d1 --- /dev/null +++ b/tinyagentos/routes/user_shares.py @@ -0,0 +1,315 @@ +from __future__ import annotations + +"""Routes for user-to-user resource sharing. + +POST /api/shares — share a resource with another user by username +GET /api/shares — list shares (direction=out → owned, direction=in → received) +POST /api/shares/{id}/accept — accept a pending share (target user only) +POST /api/shares/{id}/deny — deny a pending share (target user only) +DELETE /api/shares/{id} — revoke a share (owner or admin) + +The consent loop mirrors the external-agent consent pattern in +``agent_auth_requests.py``: on share-create a notification is raised to the +target user and a Decision record is created so the desktop consent actions +can approve / deny later. +""" + +import logging + +from fastapi import APIRouter, Depends, HTTPException, Query, Request +from pydantic import BaseModel + +from tinyagentos.auth_context import CurrentUser, current_user, require_owner_or_admin + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +# --------------------------------------------------------------------------- +# Request bodies +# --------------------------------------------------------------------------- + +VALID_RESOURCE_TYPES = ("agent", "project", "knowledge_base", "note", "file") +VALID_PERMISSIONS = ("read", "write", "admin") + + +class CreateShareRequest(BaseModel): + resource_type: str + resource_id: str + to_username: str + permission: str + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _get_user_shares_store(request: Request): + store = getattr(request.app.state, "user_shares", None) + if store is None: + raise RuntimeError("user_shares store not on app.state") + return store + + +async def user_can_access( + request: Request, resource_type: str, resource_id: str, user_id: str +) -> bool: + """Module-level helper so route consumers can check share access. + + Returns True if *user_id* has at least one active, non-expired share + for (*resource_type*, *resource_id*). Importable from + ``tinyagentos.routes.user_shares``. + """ + store = _get_user_shares_store(request) + return await store.user_can_access(resource_type, resource_id, user_id) + + +async def _find_share_by_id(request: Request, share_id: int) -> dict | None: + """Look up a share by its primary key. + + Uses ``store.get_share_by_id`` — an O(1) indexed lookup that works + regardless of ownership, status, or expiry, so admin revoke and + consent accept/deny can resolve any share. + """ + store = _get_user_shares_store(request) + return await store.get_share_by_id(share_id) + + +# --------------------------------------------------------------------------- +# Routes +# --------------------------------------------------------------------------- + +@router.post("/api/shares") +async def create_share( + request: Request, + body: CreateShareRequest, + user: CurrentUser = Depends(current_user), +): + """Share a resource with another user by username. + + Resolves *to_username* via the AuthManager → 404 if not found. + Duplicate share (same owner + resource + target + permission) is + idempotent — the store replaces the existing row. + + On create, raises a notification and a Decision record to the target + user so the desktop consent actions can approve / deny. + """ + store = _get_user_shares_store(request) + + # Resolve target user by username. + auth = getattr(request.app.state, "auth", None) + if auth is None: + raise HTTPException(status_code=500, detail="auth manager not available") + + target = auth.find_user(body.to_username) + if target is None: + raise HTTPException(status_code=404, detail=f"user '{body.to_username}' not found") + + target_user_id: str = target["id"] + + # Guard against self-share — creates a confusing UX and an unnecessary + # Decision against yourself. + if target_user_id == user.user_id: + raise HTTPException(status_code=400, detail="cannot share with yourself") + + # Validate resource_type and permission against known sets (Kilo S5). + if body.resource_type not in VALID_RESOURCE_TYPES: + raise HTTPException( + status_code=422, + detail=f"invalid resource_type '{body.resource_type}'; " + f"expected one of {VALID_RESOURCE_TYPES}", + ) + if body.permission not in VALID_PERMISSIONS: + raise HTTPException( + status_code=422, + detail=f"invalid permission '{body.permission}'; " + f"expected one of {VALID_PERMISSIONS}", + ) + + # Create (or replace) the share. The store's write lock makes this + # idempotent for concurrent same-key writes. + record = await store.add_share( + owner_user_id=user.user_id, + resource_type=body.resource_type, + resource_id=body.resource_id, + shared_with_user_id=target_user_id, + permission=body.permission, + ) + + # ------------------------------------------------------------------ + # Consent wiring — notification (same pattern as agent_auth_requests.py + # lines 204-221). Best effort: a notification failure must not fail the + # created share. + # ------------------------------------------------------------------ + notifs = getattr(request.app.state, "notifications", None) + if notifs is not None: + try: + await notifs.add( + title="Resource shared with you", + message=( + f"{user.user_id} shared {body.resource_type}/{body.resource_id} " + f"with you (permission: {body.permission})" + ), + level="info", + source="user_shares", + user_id=target_user_id, + data={ + "share_id": record["id"], + "owner_user_id": user.user_id, + "resource_type": body.resource_type, + "resource_id": body.resource_id, + "permission": body.permission, + }, + ) + except Exception: + logger.warning( + "user_shares: notification for share %s failed", + record["id"], + exc_info=True, + ) + + # ------------------------------------------------------------------ + # Consent wiring — Decision record for the target user's Decisions + # inbox so desktop consent actions (approve/deny) can act on it. + # + # NOTE: The Decision created here is intentionally NOT resolved by + # accept_share/deny_share/revoke_share. This mirrors the consent + # pattern in agent_auth_requests.py, where the Decision is an inbox + # notification that a pending consent action exists — the API + # accept/deny endpoints update the share directly. Full Decision + # lifecycle resolution (linking the Decision id back to the share + # and resolving it on consent) is deferred to a future pass. + # ------------------------------------------------------------------ + decision_store = getattr(request.app.state, "decision_store", None) + if decision_store is not None: + try: + await decision_store.create( + from_agent=user.user_id, + question=( + f"{user.user_id} shared {body.resource_type}/{body.resource_id} " + f"with you (permission: {body.permission})" + ), + type="approve_deny", + user_id=target_user_id, + context=f"Resource share from {user.user_id}", + metadata={ + "share_id": record["id"], + "owner_user_id": user.user_id, + "resource_type": body.resource_type, + "resource_id": body.resource_id, + "permission": body.permission, + }, + ) + except Exception: + logger.warning( + "user_shares: decision for share %s failed", + record["id"], + exc_info=True, + ) + + return record + + +@router.get("/api/shares") +async def list_shares( + request: Request, + direction: str = Query("out", pattern="^(out|in)$"), + user: CurrentUser = Depends(current_user), +): + """List shares for the authenticated user. + + *direction=out* (default): shares the user owns (what you've shared). + *direction=in*: shares where the user is the target (what's shared with you). + """ + store = _get_user_shares_store(request) + + if direction == "in": + return await store.list_shares_received(user.user_id) + return await store.list_shares(user.user_id) + + +@router.delete("/api/shares/{share_id}") +async def revoke_share( + request: Request, + share_id: int, + user: CurrentUser = Depends(current_user), +): + """Revoke a share by id. Owner or admin only. + + Loads the share first to obtain *owner_user_id*, then applies the + ``require_owner_or_admin`` gate against it — the admin path covers + removal of any share regardless of ownership. + """ + store = _get_user_shares_store(request) + + target = await _find_share_by_id(request, share_id) + if target is None: + raise HTTPException(status_code=404, detail="share not found") + + # Owner or admin gate — checked against the share's owner_user_id, not + # the caller's. Admin covers removal of any share. + require_owner_or_admin(user, target["owner_user_id"]) + + await store.revoke_share(share_id) + return {"status": "revoked", "share_id": share_id} + + +@router.post("/api/shares/{share_id}/accept") +async def accept_share( + request: Request, + share_id: int, + user: CurrentUser = Depends(current_user), +): + """Accept a pending share. Target user only. + + The target user (shared_with_user_id) must accept before the share + grants access. Once accepted, ``user_can_access`` returns True. + """ + store = _get_user_shares_store(request) + + target = await _find_share_by_id(request, share_id) + if target is None: + raise HTTPException(status_code=404, detail="share not found") + + if target["shared_with_user_id"] != user.user_id: + raise HTTPException(status_code=403, detail="only the target user may accept this share") + + if target.get("status") != "pending": + raise HTTPException(status_code=409, detail=f"share is already {target.get('status', 'terminal')}") + + updated = await store.accept_share(share_id) + if updated is None: + raise HTTPException(status_code=404, detail="share not found") + + return updated + + +@router.post("/api/shares/{share_id}/deny") +async def deny_share( + request: Request, + share_id: int, + user: CurrentUser = Depends(current_user), +): + """Deny a pending share. Target user only. + + The target user (shared_with_user_id) can deny to reject the share. + The share row is preserved with status='denied' for audit. + """ + store = _get_user_shares_store(request) + + target = await _find_share_by_id(request, share_id) + if target is None: + raise HTTPException(status_code=404, detail="share not found") + + if target["shared_with_user_id"] != user.user_id: + raise HTTPException(status_code=403, detail="only the target user may deny this share") + + if target.get("status") != "pending": + raise HTTPException(status_code=409, detail=f"share is already {target.get('status', 'terminal')}") + + updated = await store.deny_share(share_id) + if updated is None: + raise HTTPException(status_code=404, detail="share not found") + + return updated diff --git a/tinyagentos/user_shares_store.py b/tinyagentos/user_shares_store.py new file mode 100644 index 000000000..b8af2f9c7 --- /dev/null +++ b/tinyagentos/user_shares_store.py @@ -0,0 +1,305 @@ +from __future__ import annotations + +"""Store for user-to-user resource sharing. + +Records that a user (owner) shared a resource with another user, +including what permission was granted and under what tier. + +The Permissions app and the sharing consent loop read this table to +check whether a user may access a shared resource. ``list_active_shares`` +is the feed @taOSmd polls later. +""" + +import asyncio +from datetime import datetime, timezone +from typing import Optional + +import aiosqlite + +from tinyagentos.base_store import BaseStore + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS user_shares ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + owner_user_id TEXT NOT NULL, + resource_type TEXT NOT NULL, + resource_id TEXT NOT NULL, + shared_with_user_id TEXT NOT NULL, + permission TEXT NOT NULL, + tier TEXT NOT NULL DEFAULT 'once', -- reserved for future tier enforcement (currently unused) + granted_at TEXT NOT NULL, + expires_at TEXT, + status TEXT NOT NULL DEFAULT 'pending', + UNIQUE(owner_user_id, resource_type, resource_id, shared_with_user_id, permission) +); +""" + + +def _row_to_dict(row: aiosqlite.Row) -> dict: + return {k: row[k] for k in row.keys()} + + +class UserSharesStore(BaseStore): + """Persistent store for user-to-user resource shares.""" + + SCHEMA = SCHEMA + + # Serializes the DELETE-then-INSERT-then-SELECT in add_share. The + # 5-column UNIQUE on all-NOT-NULL columns would allow INSERT OR REPLACE, + # but we keep the explicit delete+insert+select under this lock so two + # concurrent same-key writes cannot interleave (one DELETE removing the + # other's row before its SELECT-back returns it, or the second INSERT + # hitting the unique). The lock makes each write atomic against others + # on this single connection. + _write_lock: asyncio.Lock + + async def init(self) -> None: + await super().init() + if self._db is not None: + self._db.row_factory = aiosqlite.Row + self._write_lock = asyncio.Lock() + + async def _post_init(self) -> None: + """Guarded schema upgrades for existing databases. + + Uses the PRAGMA table_info + ALTER TABLE pattern to add columns + without destructive migration, matching the approach used by + AgentGrantsStore and NotificationStore. + """ + cols = {row[1] for row in await (await self._db.execute( + "PRAGMA table_info(user_shares)")).fetchall()} + # `status` column — guarded ALTER so existing databases gain it. + # Existing rows (pre-status) are grandfathered as 'accepted' so + # previously working shares don't break on upgrade. + if "status" not in cols: + await self._db.execute( + "ALTER TABLE user_shares ADD COLUMN status TEXT NOT NULL DEFAULT 'accepted'" + ) + await self._db.commit() + + # ------------------------------------------------------------------ + # Write + # ------------------------------------------------------------------ + + async def add_share( + self, + owner_user_id: str, + resource_type: str, + resource_id: str, + shared_with_user_id: str, + permission: str, + *, + tier: str = "once", + expires_at: Optional[str] = None, + ) -> dict: + """Insert or replace a share for the exact 5-column key. + + *tier* is reserved for future tier-based enforcement and is + currently ignored (the column is kept for forward compatibility). + + Idempotent re-share: calling add_share again with the same + (owner_user_id, resource_type, resource_id, shared_with_user_id, + permission) tuple replaces the existing share rather than creating + a duplicate. The delete+insert+select runs under the write lock + for atomicity. + + If the existing share was already accepted, the re-share preserves + the original ``status`` and ``granted_at`` so an already-granted + access is not silently revoked. + """ + if self._db is None: + raise RuntimeError("UserSharesStore not initialised — call init() first") + + now = datetime.now(timezone.utc).isoformat() + async with self._write_lock: + # Fetch the existing row's status + granted_at so re-share + # preserves an already-accepted grant (Kilo W1 fix). + existing = await ( + await self._db.execute( + "SELECT status, granted_at FROM user_shares " + "WHERE owner_user_id = ? AND resource_type = ? " + "AND resource_id = ? AND shared_with_user_id = ? " + "AND permission = ?", + (owner_user_id, resource_type, resource_id, + shared_with_user_id, permission), + ) + ).fetchone() + + # Remove any existing row for the exact key first. + # + # NOTE: The DELETE+INSERT mints a new ``id`` on every + # idempotent re-share. Any Decision or notification that + # referenced the old ``id`` becomes an orphan. Switching to + # an UPDATE-in-place when the key already exists would keep + # the id stable, but the current delete+insert is simpler and + # matches the agent_grants_store add_share pattern. Full + # UPDATE-in-place id stability is deferred. + await self._db.execute( + "DELETE FROM user_shares " + "WHERE owner_user_id = ? AND resource_type = ? AND resource_id = ? " + "AND shared_with_user_id = ? AND permission = ?", + (owner_user_id, resource_type, resource_id, + shared_with_user_id, permission), + ) + + # Preserve accepted status + original granted_at on re-share; + # new shares start as 'pending' with the current timestamp. + if existing and existing["status"] == "accepted": + status_ = existing["status"] + granted_at_ = existing["granted_at"] + else: + status_ = "pending" + granted_at_ = now + + await self._db.execute( + """ + INSERT INTO user_shares + (owner_user_id, resource_type, resource_id, + shared_with_user_id, permission, granted_at, expires_at, status) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + (owner_user_id, resource_type, resource_id, + shared_with_user_id, permission, granted_at_, expires_at, status_), + ) + await self._db.commit() + row = await ( + await self._db.execute( + "SELECT * FROM user_shares " + "WHERE owner_user_id = ? AND resource_type = ? " + "AND resource_id = ? AND shared_with_user_id = ? " + "AND permission = ?", + (owner_user_id, resource_type, resource_id, + shared_with_user_id, permission), + ) + ).fetchone() + return _row_to_dict(row) # type: ignore[return-value] + + # ------------------------------------------------------------------ + # Read + # ------------------------------------------------------------------ + + async def list_shares(self, owner_user_id: str) -> list[dict]: + """Return all shares owned by *owner_user_id*, ordered by granted_at.""" + if self._db is None: + raise RuntimeError("UserSharesStore not initialised") + cursor = await self._db.execute( + "SELECT * FROM user_shares WHERE owner_user_id = ? ORDER BY granted_at", + (owner_user_id,), + ) + return [_row_to_dict(r) for r in await cursor.fetchall()] + + async def list_shares_received(self, shared_with_user_id: str) -> list[dict]: + """Return all shares where *shared_with_user_id* is the target.""" + if self._db is None: + raise RuntimeError("UserSharesStore not initialised") + cursor = await self._db.execute( + "SELECT * FROM user_shares WHERE shared_with_user_id = ? " + "ORDER BY granted_at", + (shared_with_user_id,), + ) + return [_row_to_dict(r) for r in await cursor.fetchall()] + + async def list_active_shares(self) -> list[dict]: + """Return all shares that are not yet expired. + + A share is active when ``expires_at IS NULL`` (never expires) or + ``expires_at > now``. + """ + if self._db is None: + raise RuntimeError("UserSharesStore not initialised") + now = datetime.now(timezone.utc).isoformat() + # Lexicographic ISO-8601 comparison is safe here because both + # ``now`` and any caller-set ``expires_at`` are UTC-normalised + # (datetime.now(timezone.utc).isoformat() produces a consistent + # UTC Z-suffixed string). See ``add_share`` docstring for the + # UTC contract. + cursor = await self._db.execute( + "SELECT * FROM user_shares " + "WHERE expires_at IS NULL OR expires_at > ? " + "ORDER BY owner_user_id, resource_type, resource_id", + (now,), + ) + return [_row_to_dict(r) for r in await cursor.fetchall()] + + async def revoke_share(self, share_id: int) -> None: + """Delete a share by its id. + + No error is raised if the share does not exist — the delete is + silently a no-op in that case. + """ + if self._db is None: + raise RuntimeError("UserSharesStore not initialised") + await self._db.execute( + "DELETE FROM user_shares WHERE id = ?", + (share_id,), + ) + await self._db.commit() + + async def get_share_by_id(self, share_id: int) -> dict | None: + """Return a share by its primary key, or None if not found. + + Direct ``SELECT ... WHERE id = ?`` — O(1) indexed lookup that + works regardless of ownership or expiry, so admin revoke and + consent accept/deny can resolve any share. + """ + if self._db is None: + raise RuntimeError("UserSharesStore not initialised") + cursor = await self._db.execute( + "SELECT * FROM user_shares WHERE id = ?", (share_id,) + ) + row = await cursor.fetchone() + return _row_to_dict(row) if row else None + + async def user_can_access( + self, resource_type: str, resource_id: str, user_id: str + ) -> bool: + """Return True if there is at least one active, non-expired share + for *user_id* on (*resource_type*, *resource_id*).""" + if self._db is None: + raise RuntimeError("UserSharesStore not initialised") + now = datetime.now(timezone.utc).isoformat() + cursor = await self._db.execute( + "SELECT 1 FROM user_shares " + "WHERE resource_type = ? AND resource_id = ? " + "AND shared_with_user_id = ? " + "AND status = 'accepted' " + "AND (expires_at IS NULL OR expires_at > ?) " + "LIMIT 1", + (resource_type, resource_id, user_id, now), + ) + row = await cursor.fetchone() + return row is not None + + # ------------------------------------------------------------------ + # Accept / Deny (consent gate) + # ------------------------------------------------------------------ + + async def accept_share(self, share_id: int) -> dict | None: + """Accept a pending share by id. Returns the updated row or None if not found.""" + if self._db is None: + raise RuntimeError("UserSharesStore not initialised") + await self._db.execute( + "UPDATE user_shares SET status = 'accepted' WHERE id = ? AND status = 'pending'", + (share_id,), + ) + await self._db.commit() + cursor = await self._db.execute( + "SELECT * FROM user_shares WHERE id = ?", (share_id,) + ) + row = await cursor.fetchone() + return _row_to_dict(row) if row else None + + async def deny_share(self, share_id: int) -> dict | None: + """Deny a pending share by id. Returns the updated row or None if not found.""" + if self._db is None: + raise RuntimeError("UserSharesStore not initialised") + await self._db.execute( + "UPDATE user_shares SET status = 'denied' WHERE id = ? AND status = 'pending'", + (share_id,), + ) + await self._db.commit() + cursor = await self._db.execute( + "SELECT * FROM user_shares WHERE id = ?", (share_id,) + ) + row = await cursor.fetchone() + return _row_to_dict(row) if row else None