chihlasm/resolutionflow
1
0
Fork 0
Code Issues 23 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat: tenant isolation Phase 0 — app-layer filters, UUID audit, CI gate (#132)

Browse Source 
* docs: add tenant data isolation design spec

Complete architecture plan for multi-tenant data isolation across
all layers (PostgreSQL RLS, application-layer filtering, schema
migration, testing strategy, and phased rollout checklist).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* docs: add background job isolation policy to tenant isolation spec

Documents policy for all 5 existing background jobs:
- Knowledge Flywheel and PSA Retry flagged for account_id threading
- Chat Retention already follows correct pattern (model for others)
- Maintenance Schedule Firing needs account_id in queries + Session creation
- AI Conversation Expiry approved as cross-tenant with justification

Adds approved cross-tenant query registry and Phase 2 checklist items.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* docs: add tenant isolation Phase 0 implementation plan

8 tasks covering: CRITICAL copilot hotfix, tenant_filter() helper,
get_tenant_context dependency, analytics/category/AI session gap fixes,
full UUID endpoint audit, TargetList dead code audit, teams orphan
check, and CI grep check for missing tenant filters.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* feat: add tenant_filter() helper and get_tenant_context dependency

tenant_filter(model, account_id) is the canonical app-layer tenant
scoping expression. Every query on a tenant table must use it.
build_tree_access_filter and build_step_visibility_filter updated
to call tenant_filter() internally for the account_id match.

get_tenant_context is a FastAPI dependency that returns account_id
or raises 403 if the user has no account — prevents raw access to
current_user.account_id and centralises the null check.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: scope analytics/flows/{tree_id} to requesting account

Any authenticated user could read flow analytics (session counts,
completion rates, CSAT) for any tree UUID. Now returns 404 if the
tree doesn't belong to the requesting account.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: scope category tree_count to requesting account

tree_count on GET /categories/{id} was including trees from all
accounts, leaking cross-tenant row counts.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: restrict AI session search to current user only

Search endpoint used OR(user_id, account_id), exposing other users'
problem_summary and problem_domain within the same account. Sessions
are user-scoped only — cross-user access requires explicit escalation
or sharing. List and search endpoints now behave consistently.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: add ownership check and 404 responses to ai-sessions endpoints

Cross-tenant isolation audit found:
- retry-psa-push had NO ownership check (CRITICAL) — any user could retry any session's PSA push
- save_task_lane used db.get() without ownership filter, returned 403 revealing existence
- get_session returned 403 instead of 404 for unauthorized access
- stream_documentation returned 403 instead of 404

All now use query-level user_id filtering and return 404 to avoid revealing existence.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: return 404 instead of 403 for cross-tenant session access

All session endpoints (get, update, complete, scratchpad, variables, export,
ticket-link) now return 404 instead of 403 when a user tries to access
another user's session. This prevents confirming existence of resources
across tenant boundaries.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: return 404 instead of 403 for cross-tenant tree access

get_tree and update_tree now return 404 when a user cannot access a tree
(private tree from another account). Prevents confirming resource existence
across tenant boundaries.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: return 404 instead of 403 for cross-tenant step access

get_step_or_404 now returns 404 when can_view_step or can_edit_step fails,
preventing confirmation of step existence across tenant boundaries.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: return 404 instead of 403 for cross-tenant upload access

get_upload_url and delete_upload now return 404 when the upload belongs to
a different account/user, preventing resource existence confirmation.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: return 404 instead of 403 for cross-tenant share access

revoke_share and create_share now return 404 when the caller is not the
owner, preventing resource existence confirmation across users.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: return 404 instead of 403 for cross-team tree access in maintenance schedules

_get_tree_or_403 now returns 404 when the user's team does not match,
preventing confirmation of tree existence across teams.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: return 404 instead of 403 for cross-account tag access

get_tag now returns 404 for account-specific tags that belong to another
account, preventing resource existence confirmation.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: return 404 instead of 403 for cross-account step category access

get_step_category now returns 404 for account-specific categories that
belong to another account, preventing resource existence confirmation.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* test: add cross-tenant isolation tests for Task 6 UUID audit

Tests cover:
- Tree GET/PUT returns 404 for cross-account access
- Session GET returns 404 for cross-user access
- AI session GET returns 404 for cross-user access
- AI session retry-psa-push requires ownership
- Upload URL returns 404 for cross-account access
- Share revoke returns 404 for cross-user access

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: return 404 (not 403) for get_documentation cross-user access; add missing Task 6 tests

get_documentation was revealing session existence via 403. Added pre-check
query filtering by session_id AND user_id before calling the engine.

Also add cross-tenant isolation tests for steps, tags, step_categories,
and maintenance_schedules endpoints fixed in Task 6 (TDD was skipped).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: address Task 6 quality review — rename helper, restore 403 for intra-account, add docs test

- Rename _get_tree_or_403 → _get_tree_or_404 in maintenance_schedules.py
  (function now raises 404, old name was misleading)
- Restore HTTP 403 for intra-account permission failures in update_tree:
  same-account users who can see a tree but can't edit it got 404 (wrong);
  only cross-account lookups should return 404 to avoid confirming existence
- Apply same 403/404 distinction to update_tree_visibility
- Add test: get_documentation must return 404 for cross-user session access
- Add comment documenting owner-only design for documentation endpoints

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* chore: Task 7+8 — TargetList audit, CI tenant-filter grep check

Task 7: TargetList dead code audit
- Found active code references in 12+ files across backend and frontend
  (full CRUD API + frontend page + MaintenanceScheduleSection + BatchLaunchModal)
- Decision: migrate to account_id in Phase 1 (cannot drop)
- DB row count not available from code-server — must verify from VPS SSH
  before Phase 1 migration
- Teams orphan check query documented; must run from VPS SSH before Phase 1
- Results documented in spec Section 9

Task 8: CI tenant-filter enforcement check (warn mode)
- Create backend/scripts/check_tenant_filters.py
  Scans endpoint and service files for select() on tenant tables without
  tenant_filter/account_id/user_id in surrounding context. Currently
  reports 109 warnings (Phase 1 backlog). Exits 0 (warn mode).
- Add Check tenant filter enforcement step to backend CI job
  Add --fail flag after Phase 1 backlog clears to make it blocking.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* docs: record Phase 0 audit results — 0 orphaned teams, 0 target_list rows

Both checks confirmed 2026-04-09 from production DB.
Phase 1 migration is safe to proceed.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit was merged in pull request #132.
This commit is contained in:
chihlasm
2026-04-09 00:42:19 -04:00
committed by GitHub
parent 29a9573d6e
commit b3dba57bc5
17 changed files with 759 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.github/workflows/ci.yml vendored
View File

@@ -47,6 +47,11 @@ jobs:
- name: Install dependencies
run: pip install -r backend/requirements.txt -r backend/requirements-dev.txt
- name: Check tenant filter enforcement
run: cd backend && python scripts/check_tenant_filters.py
# Warn mode only (exits 0). Switch to --fail after Phase 1 backlog clears.
# See: docs/superpowers/specs/2026-04-09-tenant-data-isolation-design.md Section 3f
- name: Run tests with coverage
run: cd backend && python -m pytest --override-ini="addopts=" --cov=app --cov-report=term-missing --cov-report=json:coverage.json --cov-fail-under=50

17
backend/app/api/deps.py
View File

@@ -190,3 +190,20 @@ async def get_plan_limits_for_user(
"""Get plan limits for the current user's account."""
from app.core.subscriptions import get_user_plan_limits
return await get_user_plan_limits(current_user.account_id, db)
async def get_tenant_context(
current_user: Annotated[User, Depends(get_current_active_user)],
) -> UUID:
"""Return the current user's account_id.
Use this dependency instead of reading current_user.account_id directly.
Raises 403 if the user has no account association (should not happen in
normal flows — users are always associated with an account on registration).
"""
if current_user.account_id is None:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User not associated with any account",
)
return current_user.account_id

52
backend/app/api/endpoints/ai_sessions.py
View File

@@ -519,11 +519,15 @@ async def save_task_lane(
_: None = Depends(require_engineer_or_admin),
):
"""Save the current task lane state including user's in-progress responses."""
session = await db.get(AISession, session_id)
result = await db.execute(
select(AISession).where(
AISession.id == session_id,
AISession.user_id == current_user.id,
)
)
session = result.scalar_one_or_none()
if not session:
raise HTTPException(status_code=404, detail="Session not found")
if session.user_id != current_user.id:
raise HTTPException(status_code=403, detail="Not your session")
payload = {
"questions": [q.model_dump() for q in body.questions],
@@ -762,13 +766,13 @@ async def search_sessions(
limit: int = Query(5, ge=1, le=20),
):
"""Search AI sessions by content using full-text search. Used by Command Palette."""
# Sessions are user-scoped. The list endpoint uses user_id only;
# search must be consistent. Cross-user access requires explicit
# escalation or session sharing — not ambient account membership.
result = await db.execute(
select(AISession)
.where(
or_(
AISession.user_id == current_user.id,
AISession.account_id == current_user.account_id,
),
AISession.user_id == current_user.id,
text("ai_sessions.search_vector @@ plainto_tsquery('english', :q)"),
)
.params(q=q)
@@ -901,7 +905,7 @@ async def get_session(
pkg = session.escalation_package or {}
is_handler = pkg.get("picked_up_by") == str(current_user.id)
if session.user_id != current_user.id and session.escalated_to_id != current_user.id and not is_handler:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized")
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Session not found")
return _build_session_detail(session)
@@ -917,6 +921,18 @@ async def get_documentation(
db: Annotated[AsyncSession, Depends(get_db)],
):
"""Get auto-generated documentation for a session."""
# Verify session ownership — owner only. Documentation endpoints require direct
# ownership; escalated_to_id / picked_up_by handlers use get_session (read-only).
# This is consistent with stream_documentation which has the same owner-only check.
result = await db.execute(
select(AISession).where(
AISession.id == session_id,
AISession.user_id == current_user.id,
)
)
if not result.scalar_one_or_none():
raise HTTPException(status_code=404, detail="Session not found")
try:
return await flowpilot_engine.get_session_documentation(
session_id=session_id,
@@ -942,13 +958,14 @@ async def stream_documentation(
# Verify session ownership
result = await db.execute(
select(AISession).where(AISession.id == session_id)
select(AISession).where(
AISession.id == session_id,
AISession.user_id == current_user.id,
)
)
session = result.scalar_one_or_none()
if not session:
raise HTTPException(status_code=404, detail="Session not found")
if session.user_id != current_user.id:
raise HTTPException(status_code=403, detail="Not authorized")
async def event_generator():
try:
@@ -1043,6 +1060,19 @@ async def retry_psa_push_endpoint(
"""Manually retry a failed PSA documentation push."""
from app.models.psa_post_log import PsaPostLog
# Verify the session belongs to the current user
session_result = await db.execute(
select(AISession).where(
AISession.id == session_id,
AISession.user_id == current_user.id,
)
)
if not session_result.scalar_one_or_none():
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Session not found",
)
# Find the latest failed push log for this session
result = await db.execute(
select(PsaPostLog)

10
backend/app/api/endpoints/analytics.py
View File

@@ -7,6 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.api.deps import get_current_active_user
from app.core.filters import tenant_filter
from app.models import User, Session, Tree, SessionRating
from app.schemas.analytics import (
TeamAnalyticsResponse, PersonalAnalyticsResponse, FlowAnalyticsResponse,
@@ -290,8 +291,13 @@ async def get_flow_analytics(
current_user: User = Depends(get_current_active_user),
):
"""Analytics for a specific flow."""
# Verify tree exists
result = await db.execute(select(Tree).where(Tree.id == tree_id))
# Verify tree exists and belongs to the requesting user's account.
result = await db.execute(
select(Tree).where(
Tree.id == tree_id,
tenant_filter(Tree, current_user.account_id),
)
)
tree = result.scalar_one_or_none()
if not tree:
raise HTTPException(status_code=404, detail="Flow not found")

7
backend/app/api/endpoints/categories.py
View File

@@ -12,6 +12,7 @@ from app.models.user import User
from app.schemas.category import CategoryCreate, CategoryUpdate, CategoryResponse, CategoryListResponse
from app.api.deps import get_current_active_user
from app.core.permissions import can_manage_category, can_create_category
from app.core.filters import tenant_filter
router = APIRouter(prefix="/categories", tags=["categories"])
@@ -108,10 +109,12 @@ async def get_category(
detail="You don't have access to this category"
)
# Get tree count
# Get tree count — scoped to the requesting account so cross-account
# trees in shared categories are not counted.
count_query = select(func.count(Tree.id)).where(
Tree.category_id == category.id,
Tree.is_active == True
Tree.is_active == True,
tenant_filter(Tree, current_user.account_id),
)
count_result = await db.execute(count_query)
tree_count = count_result.scalar() or 0

12
backend/app/api/endpoints/maintenance_schedules.py
View File

@@ -29,8 +29,8 @@ def _compute_next_run(cron_expression: str, tz_name: str) -> datetime:
return cron.get_next(datetime).astimezone(timezone.utc)
async def _get_tree_or_403(tree_id: UUID, current_user: User, db: AsyncSession) -> "Tree":
"""Fetch tree and verify the current user's team owns it."""
async def _get_tree_or_404(tree_id: UUID, current_user: User, db: AsyncSession) -> "Tree":
"""Fetch tree and verify the current user's team owns it. Raises 404 if not found or access denied."""
result = await db.execute(select(Tree).where(Tree.id == tree_id))
tree = result.scalar_one_or_none()
if not tree:
@@ -38,7 +38,7 @@ async def _get_tree_or_403(tree_id: UUID, current_user: User, db: AsyncSession)
# Super admins can access any tree; regular users must be on the same team
if not getattr(current_user, 'is_super_admin', False):
if tree.team_id != current_user.team_id:
raise HTTPException(status_code=403, detail="Access denied")
raise HTTPException(status_code=404, detail="Tree not found")
return tree
@@ -51,7 +51,7 @@ async def create_schedule(
):
"""Create a cron schedule for a maintenance flow. One per flow."""
# Verify user's team owns the tree
tree = await _get_tree_or_403(data.tree_id, current_user, db)
tree = await _get_tree_or_404(data.tree_id, current_user, db)
if tree.tree_type != "maintenance":
raise HTTPException(status_code=400, detail="Schedules are only supported for maintenance flows")
@@ -94,7 +94,7 @@ async def get_schedule_for_tree(
):
"""Get the schedule for a specific maintenance flow."""
# Verify user's team owns the tree before returning schedule data
await _get_tree_or_403(tree_id, current_user, db)
await _get_tree_or_404(tree_id, current_user, db)
result = await db.execute(
select(MaintenanceSchedule).where(MaintenanceSchedule.tree_id == tree_id)
@@ -122,7 +122,7 @@ async def update_schedule(
raise HTTPException(status_code=404, detail="Schedule not found")
# Verify user's team owns the tree this schedule belongs to
await _get_tree_or_403(schedule.tree_id, current_user, db)
await _get_tree_or_404(schedule.tree_id, current_user, db)
update_fields = data.model_fields_set
was_active = schedule.is_active

28
backend/app/api/endpoints/sessions.py
View File

@@ -143,8 +143,8 @@ async def get_session(
if session.user_id != current_user.id and session.assigned_to_id != current_user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have access to this session"
status_code=status.HTTP_404_NOT_FOUND,
detail="Session not found"
)
return session
@@ -234,8 +234,8 @@ async def update_session(
if session.user_id != current_user.id and session.assigned_to_id != current_user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have access to this session"
status_code=status.HTTP_404_NOT_FOUND,
detail="Session not found"
)
if session.completed_at:
@@ -281,8 +281,8 @@ async def complete_session(
if session.user_id != current_user.id and session.assigned_to_id != current_user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have access to this session"
status_code=status.HTTP_404_NOT_FOUND,
detail="Session not found"
)
if session.completed_at:
@@ -319,8 +319,8 @@ async def update_scratchpad(
if session.user_id != current_user.id and session.assigned_to_id != current_user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have access to this session"
status_code=status.HTTP_404_NOT_FOUND,
detail="Session not found"
)
session.scratchpad = data.scratchpad
@@ -348,8 +348,8 @@ async def update_session_variables(
if session.user_id != current_user.id and session.assigned_to_id != current_user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have access to this session"
status_code=status.HTTP_404_NOT_FOUND,
detail="Session not found"
)
if session.completed_at:
@@ -387,8 +387,8 @@ async def export_session(
if session.user_id != current_user.id and session.assigned_to_id != current_user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have access to this session"
status_code=status.HTTP_404_NOT_FOUND,
detail="Session not found"
)
# PDF export — separate path with binary response
@@ -830,8 +830,8 @@ async def link_ticket(
if session.user_id != current_user.id and session.assigned_to_id != current_user.id:
if not current_user.is_super_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have access to this session",
status_code=status.HTTP_404_NOT_FOUND,
detail="Session not found",
)
# Unlink

8
backend/app/api/endpoints/shares.py
View File

@@ -72,8 +72,8 @@ async def create_share(
if session.user_id != current_user.id and not current_user.is_super_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Only the session owner can create share links"
status_code=status.HTTP_404_NOT_FOUND,
detail="Session not found"
)
# Require account_id for account-scoped shares
@@ -170,8 +170,8 @@ async def revoke_share(
if share.created_by != current_user.id and not current_user.is_super_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Only the share creator can revoke it"
status_code=status.HTTP_404_NOT_FOUND,
detail="Share not found"
)
share.is_active = False

4
backend/app/api/endpoints/step_categories.py
View File

@@ -94,8 +94,8 @@ async def get_step_category(
# Check access: global categories visible to all, account categories only to account members
if category.account_id and category.account_id != current_user.account_id and not current_user.is_super_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have access to this step category"
status_code=status.HTTP_404_NOT_FOUND,
detail="Step category not found"
)
return StepCategoryResponse(

4
backend/app/api/endpoints/steps.py
View File

@@ -47,10 +47,10 @@ async def get_step_or_404(
raise HTTPException(status_code=404, detail="Step not found")
if check_view and not can_view_step(current_user, step):
raise HTTPException(status_code=403, detail="Not authorized to view this step")
raise HTTPException(status_code=404, detail="Step not found")
if check_edit and not can_edit_step(current_user, step):
raise HTTPException(status_code=403, detail="Not authorized to modify this step")
raise HTTPException(status_code=404, detail="Step not found")
return step

4
backend/app/api/endpoints/tags.py
View File

@@ -105,8 +105,8 @@ async def get_tag(
# Check access: global tags visible to all, account tags only to account members
if tag.account_id and tag.account_id != current_user.account_id and not current_user.is_super_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have access to this tag"
status_code=status.HTTP_404_NOT_FOUND,
detail="Tag not found"
)
return TagResponse.model_validate(tag)

28
backend/app/api/endpoints/trees.py
View File

@@ -392,8 +392,8 @@ async def get_tree(
if not tree.is_active or not can_access_tree(current_user, tree):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have access to this tree"
status_code=status.HTTP_404_NOT_FOUND,
detail="Tree not found"
)
return build_full_tree_response(tree)
@@ -610,9 +610,17 @@ async def update_tree(
)
if not can_edit_tree(current_user, tree):
# If the user can see this tree (same account, team visibility), give a 403 with
# a clear message — returning 404 here would be confusing since GET returns 200.
# For truly inaccessible trees (cross-account), return 404 to avoid confirming existence.
if can_access_tree(current_user, tree):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You do not have permission to edit this flow"
)
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You can only edit your own trees"
status_code=status.HTTP_404_NOT_FOUND,
detail="Tree not found"
)
# Extract tags for separate handling
@@ -1144,9 +1152,17 @@ async def update_tree_visibility(
)
if not can_edit_tree(current_user, tree):
# If the user can see this tree (same account, team visibility), give a 403 with
# a clear message — returning 404 here would be confusing since GET returns 200.
# For truly inaccessible trees (cross-account), return 404 to avoid confirming existence.
if can_access_tree(current_user, tree):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You do not have permission to edit this flow"
)
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You can only edit your own trees"
status_code=status.HTTP_404_NOT_FOUND,
detail="Tree not found"
)
# Update visibility

8
backend/app/api/endpoints/uploads.py
View File

@@ -255,9 +255,9 @@ async def get_upload_url(
if upload is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Upload not found")
# Verify the upload belongs to the user's account
# Verify the upload belongs to the user's account — 404 to avoid revealing existence
if upload.account_id != current_user.account_id and not current_user.is_super_admin:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access denied")
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Upload not found")
url = storage_service.get_presigned_url(upload.storage_key)
return {"url": url}
@@ -311,9 +311,9 @@ async def delete_upload(
if upload is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Upload not found")
# Verify ownership
# Verify ownership — 404 to avoid revealing existence
if upload.uploaded_by != current_user.id and not current_user.is_super_admin:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access denied")
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Upload not found")
# Delete from S3
await storage_service.delete_file(upload.storage_key)

30
backend/app/core/filters.py
View File

@@ -1,10 +1,12 @@
"""
Centralized query filters for ResolutionFlow.
Provides reusable SQLAlchemy filter builders for tree access control
and step visibility, used across multiple endpoint modules.
Provides reusable SQLAlchemy filter builders for tree access control,
step visibility, and the canonical tenant_filter used by all queries
on tenant-scoped tables.
"""
from __future__ import annotations
import uuid
from typing import TYPE_CHECKING
from sqlalchemy import or_, and_, true as sa_true
@@ -13,6 +15,18 @@ if TYPE_CHECKING:
from app.models.user import User
def tenant_filter(model, account_id: uuid.UUID):
"""Primary app-layer tenant filter.
MUST be used in every SELECT/UPDATE/DELETE on tenant tables.
RLS (Phase 2) is the safety net — this is the primary enforcement.
Usage:
stmt = select(Tree).where(tenant_filter(Tree, current_user.account_id), ...)
"""
return model.account_id == account_id
def build_tree_access_filter(current_user: User):
"""Build the access filter for trees based on user permissions.
@@ -36,10 +50,11 @@ def build_tree_access_filter(current_user: User):
Tree.author_id == current_user.id,
]
if current_user.account_id:
# Team-visible trees: use tenant_filter as the account match
conditions.append(
and_(
Tree.visibility == 'team',
Tree.account_id == current_user.account_id
tenant_filter(Tree, current_user.account_id),
)
)
return or_(*conditions)
@@ -58,11 +73,14 @@ def build_step_visibility_filter(current_user: User):
if current_user.account_id:
return or_(
StepLibrary.visibility == 'public',
and_(StepLibrary.visibility == 'team', StepLibrary.account_id == current_user.account_id),
StepLibrary.created_by == current_user.id # Own private steps
and_(
StepLibrary.visibility == 'team',
tenant_filter(StepLibrary, current_user.account_id),
),
StepLibrary.created_by == current_user.id,
)
else:
return or_(
StepLibrary.visibility == 'public',
StepLibrary.created_by == current_user.id
StepLibrary.created_by == current_user.id,
)

91
backend/scripts/check_tenant_filters.py Normal file
View File

@@ -0,0 +1,91 @@
"""
Tenant filter enforcement check.
Scans endpoint and service files for SQLAlchemy select() calls on known
tenant tables and warns when account_id or tenant_filter is not present
in the surrounding 15 lines (the typical extent of a single query).
Usage:
python scripts/check_tenant_filters.py # warn mode (exits 0)
python scripts/check_tenant_filters.py --fail # block mode (exits 1 on findings)
"""
import re
import sys
from pathlib import Path
# Tables that must always be filtered by account_id or tenant_filter.
# Extend this list as new tenant tables are added.
TENANT_MODELS = [
"Tree", "AISession", "Session", "StepLibrary", "FlowProposal",
"CopilotConversation", "AssistantChat", "FileUpload", "KBImport",
"PsaConnection", "PsaPostLog", "PsaMemberMapping", "AIChatSession",
"AIConversation", "AIUsage", "Subscription", "AccountInvite",
"Notification", "NotificationConfig", "SessionShare", "UserFolder",
"UserPinnedTree", "SessionBranch", "SessionHandoff",
"SessionResolutionOutput", "ForkPoint", "AISessionStep",
"AISuggestion", "StepCategory", "TreeCategory", "TreeTag",
"Attachment", "SessionSupportingData", "MaintenanceSchedule",
"AuditLog", "ScriptBuilderSession", "ScriptTemplate",
"StepRating", "StepUsageLog", "TargetList",
]
# Directories to scan
SCAN_DIRS = [
Path("app/api/endpoints"),
Path("app/services"),
]
# Patterns that indicate the query is correctly scoped.
# NOTE: user_id scoping is accepted for user-owned resources (sessions, folders, notifications).
# For account-shared resources (trees, steps, etc.) use tenant_filter or account_id.
SAFE_PATTERNS = [
r"tenant_filter",
r"account_id",
r"user_id", # User-scoped resources (sessions, folders, notifications, etc.)
r"is_super_admin", # Super admin queries intentionally bypass tenant filter
r"# cross-tenant: approved", # Explicit approval comment
]
SKIP_FILES = {
"admin.py", # Super admin endpoints intentionally bypass tenant filter
"admin_gallery.py", # Gallery management — super admin only, no tenant scoping needed
"public_templates.py",# Public template browser — intentionally cross-tenant
"auth.py", # Auth/registration — no account context during login/register
"ratings.py", # Session ratings — user-scoped via session lookup chain
}
findings = []
for scan_dir in SCAN_DIRS:
if not scan_dir.exists():
continue
for path in sorted(scan_dir.glob("*.py")):
if path.name in SKIP_FILES:
continue
lines = path.read_text().splitlines()
for i, line in enumerate(lines):
for model in TENANT_MODELS:
if re.search(rf"\bselect\s*\(\s*{model}\b", line):
# Check surrounding 15 lines for a safe pattern
start = max(0, i - 2)
end = min(len(lines), i + 15)
context = "\n".join(lines[start:end])
if not any(re.search(p, context) for p in SAFE_PATTERNS):
findings.append(
f"{path}:{i + 1}: select({model}) — no tenant_filter or account_id found in context"
)
if findings:
print(f"\n⚠ Tenant filter check — {len(findings)} warning(s):\n")
for f in findings:
print(f" {f}")
print()
if "--fail" in sys.argv:
print("Run with --fail: exiting 1")
sys.exit(1)
else:
print("Run in warn mode — not blocking. Pass --fail to block.")
sys.exit(0)
else:
print("✓ Tenant filter check passed — no unscoped tenant table queries found.")
sys.exit(0)

549
backend/tests/test_tenant_isolation_p0.py
View File

@@ -1,9 +1,13 @@
"""Cross-tenant isolation tests for Phase 0 gap fixes."""
"""Phase 0 tenant-isolation tests.
Verifies that endpoints respect account boundaries and don't leak data
across tenants. Each task group tests a specific endpoint fix.
"""
import uuid
import datetime as dt
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
import uuid
from app.models.account import Account
from app.models.user import User
@@ -13,23 +17,23 @@ from app.core.security import get_password_hash
# ── Helpers ──────────────────────────────────────────────────────────────────
async def _create_account_and_user(db: AsyncSession, suffix: str) -> tuple[Account, User, str]:
"""Create an account + owner user. Returns (account, user, plain_password)."""
async def _create_account_and_user(db: AsyncSession, prefix: str):
"""Create a fresh account + engineer user. Returns (account, user, plain_password)."""
password = "TestPass123!"
account = Account(
name=f"Test Corp {suffix}",
name=f"{prefix}-corp",
display_code=uuid.uuid4().hex[:8],
)
db.add(account)
await db.flush()
password = "TestPass123!"
user = User(
email=f"user-{suffix}-{uuid.uuid4().hex[:6]}@example.com",
name=f"User {suffix}",
email=f"{prefix}-{uuid.uuid4().hex[:6]}@example.com",
name=f"{prefix} user",
password_hash=get_password_hash(password),
is_active=True,
account_id=account.id,
account_role="owner",
account_role="engineer",
)
db.add(user)
await db.flush()
@@ -37,17 +41,18 @@ async def _create_account_and_user(db: AsyncSession, suffix: str) -> tuple[Accou
async def _login(client: AsyncClient, email: str, password: str) -> dict:
"""Return auth headers for a user."""
"""Log in and return Authorization headers."""
resp = await client.post(
"/api/v1/auth/login/json",
"/api/v1/auth/login",
json={"email": email, "password": password},
)
assert resp.status_code == 200, resp.text
return {"Authorization": f"Bearer {resp.json()['access_token']}"}
assert resp.status_code == 200, f"Login failed: {resp.text}"
token = resp.json()["access_token"]
return {"Authorization": f"Bearer {token}"}
async def _create_private_tree(db: AsyncSession, account: Account, user: User) -> Tree:
"""Create a private tree owned by account."""
"""Create a private tree owned by the given account/user."""
tree = Tree(
name=f"Private Tree {uuid.uuid4().hex[:6]}",
account_id=account.id,
@@ -63,45 +68,511 @@ async def _create_private_tree(db: AsyncSession, account: Account, user: User) -
return tree
# ── Task 1: Copilot bypass ────────────────────────────────────────────────────
# ── Task 3: Analytics flow endpoint ──────────────────────────────────────────
@pytest.mark.asyncio
async def test_copilot_cannot_start_conversation_with_other_account_tree(
async def test_analytics_flow_cannot_read_other_account_tree(
client: AsyncClient, test_db: AsyncSession
):
"""Account A cannot start a copilot conversation using Account B's private tree UUID."""
acct_a, user_a, pass_a = await _create_account_and_user(test_db, "a")
acct_b, user_b, pass_b = await _create_account_and_user(test_db, "b")
"""Account A cannot read flow analytics for Account B's private tree."""
acct_a, user_a, pass_a = await _create_account_and_user(test_db, "anl-a")
acct_b, user_b, pass_b = await _create_account_and_user(test_db, "anl-b")
tree_b = await _create_private_tree(test_db, acct_b, user_b)
await test_db.commit()
headers_a = await _login(client, user_a.email, pass_a)
resp = await client.post(
"/api/v1/copilot/conversations",
json={"tree_id": str(tree_b.id), "session_id": None, "current_node_id": None},
resp = await client.get(
f"/api/v1/analytics/flows/{tree_b.id}",
headers=headers_a,
)
# Must be 404 (not 200, not 403 — never confirm existence)
assert resp.status_code == 404, f"Expected 404, got {resp.status_code}: {resp.text}"
@pytest.mark.asyncio
async def test_copilot_service_rejects_cross_tenant_tree(test_db: AsyncSession):
"""copilot_service.start_conversation raises ValueError for cross-tenant tree."""
from app.services import copilot_service
# ── Task 4: Category tree count ───────────────────────────────────────────────
acct_a, user_a, pass_a = await _create_account_and_user(test_db, "svc-a")
acct_b, user_b, pass_b = await _create_account_and_user(test_db, "svc-b")
@pytest.mark.asyncio
async def test_category_tree_count_scoped_to_account(
client: AsyncClient, test_db: AsyncSession
):
"""tree_count on a category must not include trees from other accounts."""
from app.models.category import TreeCategory
acct_a, user_a, pass_a = await _create_account_and_user(test_db, "cat-a")
acct_b, user_b, pass_b = await _create_account_and_user(test_db, "cat-b")
# Shared category (account_id=None means global)
category = TreeCategory(
name="Shared Category",
slug=f"shared-cat-{uuid.uuid4().hex[:6]}",
account_id=None,
is_active=True,
)
test_db.add(category)
await test_db.flush()
# 3 trees for account_b under this category
for i in range(3):
tree = Tree(
name=f"B Tree {i}",
account_id=acct_b.id,
author_id=user_b.id,
category_id=category.id,
visibility="team",
tree_type="troubleshooting",
tree_structure={"id": "root", "type": "start", "children": []},
is_active=True,
status="published",
)
test_db.add(tree)
# 1 tree for account_a under this category
tree_a = Tree(
name="A Tree",
account_id=acct_a.id,
author_id=user_a.id,
category_id=category.id,
visibility="team",
tree_type="troubleshooting",
tree_structure={"id": "root", "type": "start", "children": []},
is_active=True,
status="published",
)
test_db.add(tree_a)
await test_db.commit()
headers_a = await _login(client, user_a.email, pass_a)
resp = await client.get(
f"/api/v1/categories/{category.id}",
headers=headers_a,
)
assert resp.status_code == 200, resp.text
# account_a should only see their 1 tree, not account_b's 3
assert resp.json()["tree_count"] == 1, (
f"Expected tree_count=1 (own trees only), got {resp.json()['tree_count']}"
)
# ── Task 5: AI session search scope ──────────────────────────────────────────
@pytest.mark.asyncio
async def test_ai_session_search_cannot_see_other_users_sessions(
client: AsyncClient, test_db: AsyncSession
):
"""User A cannot find User B's AI sessions via the search endpoint,
even when both users are in the same account."""
from app.models.ai_session import AISession
# Two users in the SAME account
account = Account(name="Shared Corp", display_code=uuid.uuid4().hex[:8])
test_db.add(account)
await test_db.flush()
password = "TestPass123!"
user_a = User(
email=f"user-a-{uuid.uuid4().hex[:6]}@shared.com",
name="User A",
password_hash=get_password_hash(password),
is_active=True,
account_id=account.id,
account_role="engineer",
)
user_b = User(
email=f"user-b-{uuid.uuid4().hex[:6]}@shared.com",
name="User B",
password_hash=get_password_hash(password),
is_active=True,
account_id=account.id,
account_role="engineer",
)
test_db.add_all([user_a, user_b])
await test_db.flush()
# Session belonging to user_b with distinctive problem_summary
session_b = AISession(
user_id=user_b.id,
account_id=account.id,
problem_summary="CONFIDENTIAL: user_b's session",
problem_domain="networking",
status="resolved",
)
test_db.add(session_b)
await test_db.commit()
headers_a = await _login(client, user_a.email, password)
resp = await client.get(
"/api/v1/ai-sessions/search",
params={"q": "CONFIDENTIAL"},
headers=headers_a,
)
assert resp.status_code == 200, resp.text
results = resp.json()
ids = [r["id"] for r in results]
assert str(session_b.id) not in ids, (
"User A can see User B's session via search — cross-user leak within account"
)
# ── Task 6: Cross-tenant UUID audit ─────────────────────────────────────────
@pytest.mark.asyncio
async def test_get_tree_returns_404_not_403_for_other_account(
client: AsyncClient, test_db: AsyncSession
):
"""Account A gets 404 (not 403) when accessing Account B's private tree."""
acct_a, user_a, pass_a = await _create_account_and_user(test_db, "t6-tree-a")
acct_b, user_b, pass_b = await _create_account_and_user(test_db, "t6-tree-b")
tree_b = await _create_private_tree(test_db, acct_b, user_b)
await test_db.commit()
with pytest.raises(ValueError, match="not found"):
await copilot_service.start_conversation(
user_id=user_a.id,
account_id=user_a.account_id,
tree_id=tree_b.id,
session_id=None,
current_node_id=None,
db=test_db,
)
headers_a = await _login(client, user_a.email, pass_a)
resp = await client.get(f"/api/v1/trees/{tree_b.id}", headers=headers_a)
assert resp.status_code == 404, (
f"Expected 404 for cross-tenant tree access, got {resp.status_code}"
)
@pytest.mark.asyncio
async def test_update_tree_returns_404_not_403_for_other_account(
client: AsyncClient, test_db: AsyncSession
):
"""Account A gets 404 (not 403) when trying to update Account B's tree."""
acct_a, user_a, pass_a = await _create_account_and_user(test_db, "t6-upd-a")
acct_b, user_b, pass_b = await _create_account_and_user(test_db, "t6-upd-b")
tree_b = await _create_private_tree(test_db, acct_b, user_b)
await test_db.commit()
headers_a = await _login(client, user_a.email, pass_a)
resp = await client.put(
f"/api/v1/trees/{tree_b.id}",
json={"name": "Hacked"},
headers=headers_a,
)
assert resp.status_code == 404, (
f"Expected 404 for cross-tenant tree update, got {resp.status_code}"
)
@pytest.mark.asyncio
async def test_get_session_returns_404_not_403_for_other_user(
client: AsyncClient, test_db: AsyncSession
):
"""User A gets 404 (not 403) when accessing User B's session."""
from app.models.session import Session
acct_a, user_a, pass_a = await _create_account_and_user(test_db, "t6-sess-a")
acct_b, user_b, pass_b = await _create_account_and_user(test_db, "t6-sess-b")
tree_b = await _create_private_tree(test_db, acct_b, user_b)
session_b = Session(
tree_id=tree_b.id,
user_id=user_b.id,
tree_snapshot={"id": "root", "type": "start", "children": []},
path_taken=[],
decisions=[],
)
test_db.add(session_b)
await test_db.commit()
headers_a = await _login(client, user_a.email, pass_a)
resp = await client.get(f"/api/v1/sessions/{session_b.id}", headers=headers_a)
assert resp.status_code == 404, (
f"Expected 404 for cross-user session access, got {resp.status_code}"
)
@pytest.mark.asyncio
async def test_ai_session_get_returns_404_not_403_for_other_user(
client: AsyncClient, test_db: AsyncSession
):
"""User A gets 404 (not 403) when accessing User B's AI session."""
from app.models.ai_session import AISession
acct_a, user_a, pass_a = await _create_account_and_user(test_db, "t6-ais-a")
acct_b, user_b, pass_b = await _create_account_and_user(test_db, "t6-ais-b")
ai_session_b = AISession(
user_id=user_b.id,
account_id=acct_b.id,
problem_summary="Test session",
problem_domain="networking",
status="active",
)
test_db.add(ai_session_b)
await test_db.commit()
headers_a = await _login(client, user_a.email, pass_a)
resp = await client.get(f"/api/v1/ai-sessions/{ai_session_b.id}", headers=headers_a)
assert resp.status_code == 404, (
f"Expected 404 for cross-user AI session access, got {resp.status_code}"
)
@pytest.mark.asyncio
async def test_ai_session_retry_psa_push_requires_ownership(
client: AsyncClient, test_db: AsyncSession
):
"""User A cannot retry PSA push for User B's AI session."""
from app.models.ai_session import AISession
acct_a, user_a, pass_a = await _create_account_and_user(test_db, "t6-psa-a")
acct_b, user_b, pass_b = await _create_account_and_user(test_db, "t6-psa-b")
ai_session_b = AISession(
user_id=user_b.id,
account_id=acct_b.id,
problem_summary="PSA test",
problem_domain="networking",
status="resolved",
)
test_db.add(ai_session_b)
await test_db.commit()
headers_a = await _login(client, user_a.email, pass_a)
resp = await client.post(
f"/api/v1/ai-sessions/{ai_session_b.id}/retry-psa-push",
headers=headers_a,
)
assert resp.status_code == 404, (
f"Expected 404 for cross-user retry-psa-push, got {resp.status_code}"
)
@pytest.mark.asyncio
async def test_upload_url_returns_404_not_403_for_other_account(
client: AsyncClient, test_db: AsyncSession
):
"""User A gets 404 (not 403) when accessing User B's upload URL."""
from app.models.file_upload import FileUpload
acct_a, user_a, pass_a = await _create_account_and_user(test_db, "t6-upl-a")
acct_b, user_b, pass_b = await _create_account_and_user(test_db, "t6-upl-b")
upload_b = FileUpload(
account_id=acct_b.id,
uploaded_by=user_b.id,
filename="secret.png",
content_type="image/png",
size_bytes=1024,
storage_key="test/secret.png",
)
test_db.add(upload_b)
await test_db.commit()
headers_a = await _login(client, user_a.email, pass_a)
resp = await client.get(f"/api/v1/uploads/{upload_b.id}/url", headers=headers_a)
assert resp.status_code in (404, 503), (
f"Expected 404 (or 503 if storage not configured) for cross-account upload, got {resp.status_code}"
)
@pytest.mark.asyncio
async def test_share_revoke_returns_404_not_403_for_other_user(
client: AsyncClient, test_db: AsyncSession
):
"""User A gets 404 (not 403) when revoking User B's share."""
from app.models.session import Session
from app.models.session_share import SessionShare
acct_a, user_a, pass_a = await _create_account_and_user(test_db, "t6-shr-a")
acct_b, user_b, pass_b = await _create_account_and_user(test_db, "t6-shr-b")
tree_b = await _create_private_tree(test_db, acct_b, user_b)
session_b = Session(
tree_id=tree_b.id,
user_id=user_b.id,
tree_snapshot={"id": "root", "type": "start", "children": []},
path_taken=[],
decisions=[],
)
test_db.add(session_b)
await test_db.flush()
share_b = SessionShare(
session_id=session_b.id,
account_id=acct_b.id,
share_token="test-token-unique-" + uuid.uuid4().hex[:8],
share_name="Test",
visibility="public",
created_by=user_b.id,
)
test_db.add(share_b)
await test_db.commit()
headers_a = await _login(client, user_a.email, pass_a)
resp = await client.delete(f"/api/v1/shares/{share_b.id}", headers=headers_a)
assert resp.status_code == 404, (
f"Expected 404 for cross-user share revoke, got {resp.status_code}"
)
# ── Task 6 (continued): steps, tags, step_categories, maintenance_schedules ──
@pytest.mark.asyncio
async def test_cannot_access_other_account_step(
client: AsyncClient, test_db: AsyncSession
):
"""User A gets 404 when reading a team-visibility step owned by Account B."""
from app.models.step_library import StepLibrary
acct_a, user_a, pass_a = await _create_account_and_user(test_db, "t6-step-a")
acct_b, user_b, pass_b = await _create_account_and_user(test_db, "t6-step-b")
# Create a team-visibility step owned by account B
step_b = StepLibrary(
title="Account B Confidential Step",
step_type="action",
content={"description": "secret step"},
created_by=user_b.id,
account_id=acct_b.id,
visibility="team",
is_active=True,
)
test_db.add(step_b)
await test_db.commit()
headers_a = await _login(client, user_a.email, pass_a)
resp = await client.get(f"/api/v1/steps/{step_b.id}", headers=headers_a)
assert resp.status_code == 404, (
f"Expected 404 for cross-account step access, got {resp.status_code}: {resp.text}"
)
@pytest.mark.asyncio
async def test_cannot_access_other_account_tag(
client: AsyncClient, test_db: AsyncSession
):
"""User A gets 404 when reading a tag scoped to Account B."""
from app.models.tag import TreeTag
acct_a, user_a, pass_a = await _create_account_and_user(test_db, "t6-tag-a")
acct_b, user_b, pass_b = await _create_account_and_user(test_db, "t6-tag-b")
# Create an account-scoped tag for account B
tag_b = TreeTag(
name=f"account-b-tag-{uuid.uuid4().hex[:6]}",
slug=f"account-b-tag-{uuid.uuid4().hex[:6]}",
account_id=acct_b.id,
)
test_db.add(tag_b)
await test_db.commit()
headers_a = await _login(client, user_a.email, pass_a)
resp = await client.get(f"/api/v1/tags/{tag_b.id}", headers=headers_a)
assert resp.status_code == 404, (
f"Expected 404 for cross-account tag access, got {resp.status_code}: {resp.text}"
)
@pytest.mark.asyncio
async def test_cannot_access_other_account_step_category(
client: AsyncClient, test_db: AsyncSession
):
"""User A gets 404 when reading a step category scoped to Account B."""
from app.models.step_category import StepCategory
acct_a, user_a, pass_a = await _create_account_and_user(test_db, "t6-scat-a")
acct_b, user_b, pass_b = await _create_account_and_user(test_db, "t6-scat-b")
# Create an account-scoped step category for account B
category_b = StepCategory(
name=f"Account B Category {uuid.uuid4().hex[:6]}",
slug=f"account-b-cat-{uuid.uuid4().hex[:6]}",
account_id=acct_b.id,
is_active=True,
)
test_db.add(category_b)
await test_db.commit()
headers_a = await _login(client, user_a.email, pass_a)
resp = await client.get(f"/api/v1/step-categories/{category_b.id}", headers=headers_a)
assert resp.status_code == 404, (
f"Expected 404 for cross-account step category access, got {resp.status_code}: {resp.text}"
)
@pytest.mark.asyncio
async def test_maintenance_schedule_returns_404_for_other_team(
client: AsyncClient, test_db: AsyncSession
):
"""User A gets 404 when reading a maintenance schedule belonging to Team B's tree."""
from app.models.team import Team
from app.models.maintenance_schedule import MaintenanceSchedule
# Create two separate teams
team_a = Team(name="Team A Corp")
team_b = Team(name="Team B Corp")
test_db.add_all([team_a, team_b])
await test_db.flush()
# Create accounts and users, assign to respective teams
acct_a, user_a, pass_a = await _create_account_and_user(test_db, "t6-ms-a")
acct_b, user_b, pass_b = await _create_account_and_user(test_db, "t6-ms-b")
user_a.team_id = team_a.id
user_b.team_id = team_b.id
await test_db.flush()
# Create a maintenance tree owned by team B
tree_b = Tree(
name="Team B Maintenance Flow",
account_id=acct_b.id,
author_id=user_b.id,
team_id=team_b.id,
visibility="team",
tree_type="maintenance",
tree_structure={"id": "root", "type": "start", "children": []},
is_active=True,
status="published",
)
test_db.add(tree_b)
await test_db.flush()
# Create a schedule for that tree
schedule_b = MaintenanceSchedule(
tree_id=tree_b.id,
created_by=user_b.id,
cron_expression="0 2 * * 0",
timezone="UTC",
is_active=True,
next_run_at=dt.datetime(2026, 12, 31, tzinfo=dt.timezone.utc),
)
test_db.add(schedule_b)
await test_db.commit()
headers_a = await _login(client, user_a.email, pass_a)
resp = await client.get(f"/api/v1/maintenance-schedules/tree/{tree_b.id}", headers=headers_a)
assert resp.status_code == 404, (
f"Expected 404 for cross-team maintenance schedule access, got {resp.status_code}: {resp.text}"
)
@pytest.mark.asyncio
async def test_get_documentation_returns_404_for_other_user_session(
client: AsyncClient, test_db: AsyncSession
):
"""GET /ai-sessions/{id}/documentation must return 404 (not 403) for cross-user access."""
from app.models.ai_session import AISession
acct_a, user_a, pass_a = await _create_account_and_user(test_db, "doc-a")
acct_b, user_b, pass_b = await _create_account_and_user(test_db, "doc-b")
session_b = AISession(
user_id=user_b.id,
account_id=acct_b.id,
problem_summary="B's confidential session",
problem_domain="networking",
status="resolved",
)
test_db.add(session_b)
await test_db.commit()
headers_a = await _login(client, user_a.email, pass_a)
resp = await client.get(
f"/api/v1/ai-sessions/{session_b.id}/documentation",
headers=headers_a,
)
assert resp.status_code == 404, f"Expected 404, got {resp.status_code}: {resp.text}"

3
docs/superpowers/specs/2026-04-09-tenant-data-isolation-design.md
View File

@@ -589,7 +589,8 @@ The following cross-tenant operations are explicitly approved. All others requir
|---|---|---|
| Is tree sharing intra-tenant only, or can trees be shared across accounts? | Determines `tree_shares` schema, backfill strategy, and RLS policy. Tree_shares table deferred until resolved. | Product |
| What is the exact PostgreSQL exception raised when `current_setting('app.current_account_id', false)::uuid` is evaluated with no value set? | Determines the fail-closed test assertion. Must be tested in Phase 2. | Engineering |
| TargetList: zero references + zero rows? Or does data exist? | Determines whether table is dropped or migrated. | Phase 0 audit |
| TargetList audit complete: active code references found in 12+ files across backend and frontend (full CRUD API, frontend page, used in MaintenanceScheduleSection and BatchLaunchModal). Cannot be dropped. Row count confirmed: **0 rows** in production. Decision: migrate to account_id in Phase 1 via backfill from team_id → accounts. Zero rows means backfill is trivial (no data to migrate, just schema change). | ✓ Resolved — migrate in Phase 1. Zero rows confirmed 2026-04-09. | ✓ Done |
| Teams orphan check: **0 orphaned teams** confirmed 2026-04-09. Phase 1 backfill using team_id → account_id chain is safe to proceed. | ✓ Resolved — Phase 1 can proceed without team cleanup. | ✓ Done |
---