leocrm-bot
461 lines
16 KiB
Python
461 lines
16 KiB
Python
"""Tests for Permissions plugin — permissions, share links, public access, 403 enforcement."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
|
|
|
|
from app.core.db import close_engine, reset_engine_for_testing
|
|
from app.core.service_container import get_container
|
|
from app.main import create_app
|
|
from app.plugins.builtins.permissions import PermissionsPlugin
|
|
from app.plugins.registry import reset_registry_for_testing
|
|
from app.services.plugin_service import reset_plugin_service_for_testing
|
|
from tests.conftest import ORIGIN_HEADER, login_client, seed_tenant_and_users
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def plugin_app(engine: AsyncEngine, redis_client):
|
|
"""FastAPI app with permissions plugin registered."""
|
|
reset_engine_for_testing(engine)
|
|
app = create_app()
|
|
|
|
registry = reset_registry_for_testing()
|
|
registry.initialize(engine, app)
|
|
|
|
container = get_container()
|
|
await container.initialize()
|
|
|
|
registry.register_plugin(PermissionsPlugin())
|
|
reset_plugin_service_for_testing(registry)
|
|
|
|
yield app
|
|
await close_engine()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def plugin_client(plugin_app) -> AsyncClient:
|
|
transport = ASGITransport(app=plugin_app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as c:
|
|
yield c
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def authed_client(plugin_client: AsyncClient, db_session: AsyncSession) -> AsyncClient:
|
|
"""Authenticated admin client with seeded data."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
await login_client(plugin_client, "admin@tenanta.com")
|
|
resp = await plugin_client.post("/api/v1/plugins/permissions/install", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
resp = await plugin_client.post("/api/v1/plugins/permissions/activate", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
return plugin_client, seed
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_permissions_empty(authed_client: AsyncClient):
|
|
"""AC1: GET /api/v1/dms/files/{id}/permissions → 200 + permission list."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
resp = await client.get(f"/api/v1/dms/files/{file_id}/permissions", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
assert resp.json() == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_grant_permission(authed_client: AsyncClient):
|
|
"""POST /api/v1/dms/files/{id}/permissions → 201, permission granted."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
user_id = str(seed["admin_a"].id)
|
|
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/permissions",
|
|
json={"user_id": user_id, "access_level": "read"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["user_id"] == user_id
|
|
assert data["access_level"] == "read"
|
|
|
|
# List permissions should show it
|
|
resp = await client.get(f"/api/v1/dms/files/{file_id}/permissions", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_revoke_permission(authed_client: AsyncClient):
|
|
"""DELETE /api/v1/dms/files/{id}/permissions/{user_id} → 204."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
user_id = str(seed["admin_a"].id)
|
|
|
|
# Grant first
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/permissions",
|
|
json={"user_id": user_id, "access_level": "write"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
|
|
# Revoke
|
|
resp = await client.delete(
|
|
f"/api/v1/dms/files/{file_id}/permissions/{user_id}",
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
# List should be empty
|
|
resp = await client.get(f"/api/v1/dms/files/{file_id}/permissions", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
assert resp.json() == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_share_link(authed_client: AsyncClient):
|
|
"""AC4: POST /api/v1/dms/files/{id}/share-link → 200 + public token URL."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/share-link",
|
|
json={"access_level": "download"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "token" in data
|
|
assert "public_url" in data
|
|
assert data["has_password"] is False
|
|
assert data["file_id"] == file_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_share_link_with_password(authed_client: AsyncClient):
|
|
"""Share link with password — POST verify with correct password succeeds."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/share-link",
|
|
json={"password": "Secret123", "access_level": "download"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["has_password"] is True
|
|
token = data["token"]
|
|
|
|
# GET without password → 401
|
|
resp = await client.get(f"/api/public/share/{token}")
|
|
assert resp.status_code == 401
|
|
|
|
# POST with wrong password → 403
|
|
resp = await client.post(
|
|
f"/api/public/share/{token}",
|
|
json={"password": "WrongPass"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
# POST with correct password → 200
|
|
resp = await client.post(
|
|
f"/api/public/share/{token}",
|
|
json={"password": "Secret123"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["file_id"] == file_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_expired_share_link(authed_client: AsyncClient):
|
|
"""AC5: GET /api/public/share/{token} with expired link → 410."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
|
|
# Create link with expiry in the past
|
|
past = datetime.now(UTC) - timedelta(hours=1)
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/share-link",
|
|
json={"expires_at": past.isoformat(), "access_level": "download"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
token = resp.json()["token"]
|
|
|
|
# GET → 410 Gone
|
|
resp = await client.get(f"/api/public/share/{token}")
|
|
assert resp.status_code == 410
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_public_share_no_password(authed_client: AsyncClient):
|
|
"""Public share link without password — GET returns file info."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/share-link",
|
|
json={"access_level": "preview"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
token = resp.json()["token"]
|
|
|
|
# Public GET — no auth, no Origin header needed
|
|
resp = await client.get(f"/api/public/share/{token}")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["file_id"] == file_id
|
|
assert data["access_level"] == "preview"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_revoke_share_link(authed_client: AsyncClient):
|
|
"""DELETE /api/v1/dms/share-links/{id} → 204, link revoked."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/share-link",
|
|
json={"access_level": "download"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
link_id = resp.json()["id"]
|
|
|
|
resp = await client.delete(f"/api/v1/dms/share-links/{link_id}", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 204
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_permission_403_for_unauthorized_user(
|
|
plugin_client: AsyncClient, db_session: AsyncSession
|
|
):
|
|
"""AC14: Folder permissions enforced: user without read → 403.
|
|
|
|
The permissions plugin does not intercept file access (no DMS file system yet).
|
|
We test the permission check helper directly: a user with no permission record
|
|
gets denied (False), which translates to 403 at the route layer.
|
|
"""
|
|
from app.core.db import get_session_factory
|
|
from app.plugins.builtins.permissions.routes import check_user_file_permission
|
|
|
|
seed = await seed_tenant_and_users(db_session)
|
|
await login_client(plugin_client, "admin@tenanta.com")
|
|
|
|
# Install + activate permissions plugin
|
|
reset_registry_for_testing()
|
|
# We need to set up the registry properly
|
|
# Since plugin_client fixture wasn't used, manually install
|
|
resp = await plugin_client.post("/api/v1/plugins/permissions/install", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
resp = await plugin_client.post("/api/v1/plugins/permissions/activate", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
|
|
file_id = uuid.uuid4()
|
|
tenant_id = seed["tenant_a"].id
|
|
viewer_id = seed["viewer_a"].id
|
|
admin_id = seed["admin_a"].id
|
|
|
|
# No permission record for viewer → check returns False
|
|
factory = get_session_factory()
|
|
async with factory() as session:
|
|
has_perm = await check_user_file_permission(session, tenant_id, file_id, viewer_id, "read")
|
|
assert has_perm is False
|
|
|
|
# Grant read to admin
|
|
from app.plugins.builtins.permissions.models import Permission
|
|
|
|
perm = Permission(
|
|
tenant_id=tenant_id,
|
|
file_id=file_id,
|
|
user_id=admin_id,
|
|
access_level="read",
|
|
)
|
|
session.add(perm)
|
|
await session.flush()
|
|
|
|
has_perm = await check_user_file_permission(session, tenant_id, file_id, admin_id, "read")
|
|
assert has_perm is True
|
|
|
|
# Viewer still has no permission
|
|
has_perm = await check_user_file_permission(session, tenant_id, file_id, viewer_id, "read")
|
|
assert has_perm is False
|
|
await session.rollback()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_permissions_invalid_file_id(authed_client: AsyncClient):
|
|
"""GET /api/v1/dms/files/{invalid}/permissions → 400."""
|
|
client, seed = authed_client
|
|
resp = await client.get("/api/v1/dms/files/bad-uuid/permissions", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_grant_permission_duplicate(authed_client: AsyncClient):
|
|
"""POST /api/v1/dms/files/{id}/permissions twice → 409."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
user_id = str(seed["admin_a"].id)
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/permissions",
|
|
json={"user_id": user_id, "access_level": "read"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/permissions",
|
|
json={"user_id": user_id, "access_level": "read"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_grant_permission_invalid_ids(authed_client: AsyncClient):
|
|
"""POST /api/v1/dms/files/{invalid}/permissions → 400."""
|
|
client, seed = authed_client
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/bad-uuid/permissions",
|
|
json={"user_id": str(uuid.uuid4()), "access_level": "read"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_grant_permission_with_group(authed_client: AsyncClient):
|
|
"""POST /api/v1/dms/files/{id}/permissions with group_id → 201."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
user_id = str(seed["admin_a"].id)
|
|
group_id = str(uuid.uuid4())
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/permissions",
|
|
json={"user_id": user_id, "group_id": group_id, "access_level": "read"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
assert resp.json()["group_id"] == group_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_revoke_permission_not_found(authed_client: AsyncClient):
|
|
"""DELETE /api/v1/dms/files/{id}/permissions/{user_id} with no perms → 404."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
user_id = str(seed["admin_a"].id)
|
|
resp = await client.delete(
|
|
f"/api/v1/dms/files/{file_id}/permissions/{user_id}",
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_revoke_permission_invalid_ids(authed_client: AsyncClient):
|
|
"""DELETE /api/v1/dms/files/{invalid}/permissions/{user_id} → 400."""
|
|
client, seed = authed_client
|
|
user_id = str(seed["admin_a"].id)
|
|
resp = await client.delete(
|
|
f"/api/v1/dms/files/bad-uuid/permissions/{user_id}",
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_share_link_invalid_file_id(authed_client: AsyncClient):
|
|
"""POST /api/v1/dms/files/{invalid}/share-link → 400."""
|
|
client, seed = authed_client
|
|
resp = await client.post(
|
|
"/api/v1/dms/files/bad-uuid/share-link",
|
|
json={"access_level": "download"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_revoke_share_link_not_found(authed_client: AsyncClient):
|
|
"""DELETE /api/v1/dms/share-links/{nonexistent} → 404."""
|
|
client, seed = authed_client
|
|
resp = await client.delete(f"/api/v1/dms/share-links/{uuid.uuid4()}", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_revoke_share_link_invalid_id(authed_client: AsyncClient):
|
|
"""DELETE /api/v1/dms/share-links/{invalid} → 400."""
|
|
client, seed = authed_client
|
|
resp = await client.delete("/api/v1/dms/share-links/bad-uuid", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_public_access_not_found(authed_client: AsyncClient):
|
|
"""GET /api/public/share/{nonexistent} → 404."""
|
|
client, seed = authed_client
|
|
resp = await client.get("/api/public/share/nonexistent-token-xyz")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_public_access_post_not_found(authed_client: AsyncClient):
|
|
"""POST /api/public/share/{nonexistent} → 404."""
|
|
client, seed = authed_client
|
|
resp = await client.post(
|
|
"/api/public/share/nonexistent-token-xyz",
|
|
json={"password": "test"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_public_access_post_expired(authed_client: AsyncClient):
|
|
"""POST /api/public/share/{token} with expired link → 410."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
past = datetime.now(UTC) - timedelta(hours=1)
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/share-link",
|
|
json={"expires_at": past.isoformat(), "access_level": "download"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
token = resp.json()["token"]
|
|
resp = await client.post(
|
|
f"/api/public/share/{token}",
|
|
json={"password": "test"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 410
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_public_access_post_no_password_required(authed_client: AsyncClient):
|
|
"""POST /api/public/share/{token} for link without password → 200."""
|
|
client, seed = authed_client
|
|
file_id = str(uuid.uuid4())
|
|
resp = await client.post(
|
|
f"/api/v1/dms/files/{file_id}/share-link",
|
|
json={"access_level": "preview"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
token = resp.json()["token"]
|
|
resp = await client.post(
|
|
f"/api/public/share/{token}",
|
|
json={},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["has_password"] is False
|