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>
This commit is contained in:
chihlasm
2026-04-09 04:12:59 +00:00
parent 8a4c870cd9
commit ff53ea7da9
2 changed files with 152 additions and 0 deletions

View File

@@ -921,6 +921,16 @@ async def get_documentation(
db: Annotated[AsyncSession, Depends(get_db)], db: Annotated[AsyncSession, Depends(get_db)],
): ):
"""Get auto-generated documentation for a session.""" """Get auto-generated documentation for a session."""
# Verify session ownership — return 404 (not 403) to avoid confirming existence.
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: try:
return await flowpilot_engine.get_session_documentation( return await flowpilot_engine.get_session_documentation(
session_id=session_id, session_id=session_id,

View File

@@ -4,6 +4,7 @@ Verifies that endpoints respect account boundaries and don't leak data
across tenants. Each task group tests a specific endpoint fix. across tenants. Each task group tests a specific endpoint fix.
""" """
import uuid import uuid
import datetime as dt
import pytest import pytest
from httpx import AsyncClient from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
@@ -406,3 +407,144 @@ async def test_share_revoke_returns_404_not_403_for_other_user(
assert resp.status_code == 404, ( assert resp.status_code == 404, (
f"Expected 404 for cross-user share revoke, got {resp.status_code}" 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}"
)