feat: v6.1.0 granular permissions (page/collection/property ACL + groups + audit)
FlowDeck CI / lint (push) Failing after 1m8s
FlowDeck CI / test (push) Failing after 8m5s
FlowDeck CI / docker (push) Skipped

- Migration 18: 6 tables + 3 colonnes permission_type + indexes
- PermissionManager: heritage page->collection->workspace, least privilege, groups, cache 60s
- API /api/v2: pages/collections/properties/groups/users/audit (401/403/404/400)
- Guards board.py + collections.py (404/403, admin/owner bypass)
- Tests 21/21 (inherit/restricted/private, grant, revoke, batch, group, audit)
- Docs + ROADMAP + CHANGELOG + VERSION 6.1.0
This commit is contained in:
2026-09-19 22:54:16 -04:00
parent d707a6850a
commit 13d5f8625a
11 changed files with 2215 additions and 14 deletions
+538
View File
@@ -0,0 +1,538 @@
"""FlowDeck — v6.0.0 Granular permissions tests.
Covers the page / collection / property ACL: permission types (inherit,
restricted, private), explicit grants (user + group), inheritance, property
visibility, user groups and the audit log.
"""
from app.auth.session import SessionManager
# ═══════════════ helpers ═══════════════
def _token(user_id, login="user", is_admin=0, full_name="User"):
return SessionManager.create_session(
{"id": user_id, "login": login, "full_name": full_name, "is_admin": is_admin}
)
def _as(client, user_id, login="user", is_admin=0):
client.cookies.set("flowdeck_session", _token(user_id, login, is_admin))
def _insert_user(login, email, full_name):
from app.db import get_conn
with get_conn() as conn:
cur = conn.execute(
"INSERT INTO users (login, email, full_name, is_active) VALUES (?,?,?,1)",
(login, email, full_name),
)
conn.commit()
return cur.lastrowid
def _insert_workspace(owner_id, name="WS"):
from app.db import get_conn
with get_conn() as conn:
cur = conn.execute(
"INSERT INTO workspaces (name, owner_id) VALUES (?,?)", (name, owner_id)
)
conn.commit()
return cur.lastrowid
def _insert_page(workspace_id, title="Page", collection_id=None):
from app.db import get_conn
with get_conn() as conn:
cur = conn.execute(
"INSERT INTO pages (workspace, workspace_id, title, content, content_format, "
"parent_section, collection_id) VALUES (?,?,?,'','blocks','Private',?)",
("ws", workspace_id, title, collection_id),
)
conn.commit()
return cur.lastrowid
def _insert_collection(workspace_id, name="DB"):
from app.db import get_conn
with get_conn() as conn:
cur = conn.execute(
"INSERT INTO collections (name, schema_json, workspace_id) VALUES (?,'[]',?)",
(name, workspace_id),
)
conn.commit()
return cur.lastrowid
def _insert_property(collection_id, name="Field"):
from app.db import get_conn
with get_conn() as conn:
cur = conn.execute(
"INSERT INTO collection_properties (collection_id, name, prop_type, position) "
"VALUES (?,?,'text',0)",
(collection_id, name),
)
conn.commit()
return cur.lastrowid
def _set_page_type(page_id, ptype):
from app.db import get_conn
with get_conn() as conn:
conn.execute("UPDATE pages SET permission_type=? WHERE id=?", (ptype, page_id))
conn.commit()
def _set_collection_type(collection_id, ptype):
from app.db import get_conn
with get_conn() as conn:
conn.execute(
"UPDATE collections SET permission_type=? WHERE id=?", (ptype, collection_id)
)
conn.commit()
def _scope(owner_id, member_id=None):
"""Return (workspace_id, page_id, owner_id, member_id) with a fresh setup."""
ws = _insert_workspace(owner_id)
page = _insert_page(ws)
if member_id:
from app.db import get_conn
with get_conn() as conn:
conn.execute(
"INSERT INTO workspace_members (workspace_id, user_id, role) VALUES (?,?,'viewer')",
(ws, member_id),
)
conn.commit()
return ws, page, owner_id, member_id
# ═══════════════ page permission types ═══════════════
def test_inherit_allows_workspace_viewer(client):
owner = _insert_user("owner", "[email protected]", "Owner")
member = _insert_user("member", "[email protected]", "Member")
_, page, _, _ = _scope(owner, member)
_as(client, member, "member")
r = client.get(f"/board/api/pages/{page}")
print("DEBUG body:", r.status_code, r.text)
assert r.status_code == 200, r.text
# Inherited viewer cannot edit.
resp = client.put(f"/board/api/pages/{page}?title=Sneak")
assert resp.status_code == 403, resp.text
def test_restricted_page_hidden_for_non_grantees(client):
owner = _insert_user("owner", "[email protected]", "Owner")
member = _insert_user("member", "[email protected]", "Member")
ws, page, _, _ = _scope(owner, member)
_set_page_type(page, "restricted")
# Owner still sees it.
_as(client, owner, "owner")
assert client.get(f"/board/api/pages/{page}").status_code == 200
# Workspace member WITHOUT a grant → hidden (404).
_as(client, member, "member")
assert client.get(f"/board/api/pages/{page}").status_code == 404
assert client.put(f"/board/api/pages/{page}?title=x").status_code == 404
def test_grant_viewer_unlocks_restricted_page(client):
owner = _insert_user("owner", "[email protected]", "Owner")
member = _insert_user("member", "[email protected]", "Member")
_, page, _, _ = _scope(owner, member)
_set_page_type(page, "restricted")
_as(client, owner, "owner")
grant = client.post(
f"/api/v2/pages/{page}/permissions",
json={"user_id": member, "role": "viewer"},
)
assert grant.status_code == 200, grant.text
_as(client, member, "member")
assert client.get(f"/board/api/pages/{page}").status_code == 200
# Viewer role cannot edit.
assert client.put(f"/board/api/pages/{page}?title=x").status_code == 403
def test_grant_editor_allows_edit_but_not_manage(client):
owner = _insert_user("owner", "[email protected]", "Owner")
member = _insert_user("member", "[email protected]", "Member")
_, page, _, _ = _scope(owner, member)
_as(client, owner, "owner")
client.post(
f"/api/v2/pages/{page}/permissions",
json={"user_id": member, "role": "editor"},
)
_as(client, member, "member")
assert client.get(f"/board/api/pages/{page}").status_code == 200
assert client.put(f"/board/api/pages/{page}?title=Edited").status_code == 200
# Editor is not an owner → cannot grant more permissions.
resp = client.post(
f"/api/v2/pages/{page}/permissions",
json={"user_id": owner, "role": "viewer"},
)
assert resp.status_code == 403
def test_revoke_removes_access(client):
owner = _insert_user("owner", "[email protected]", "Owner")
member = _insert_user("member", "[email protected]", "Member")
_, page, _, _ = _scope(owner, member)
_set_page_type(page, "restricted")
_as(client, owner, "owner")
grant = client.post(
f"/api/v2/pages/{page}/permissions",
json={"user_id": member, "role": "viewer"},
)
perm_id = grant.json()["id"]
revoke = client.delete(f"/api/v2/pages/{page}/permissions/{perm_id}")
assert revoke.status_code == 200
_as(client, member, "member")
assert client.get(f"/board/api/pages/{page}").status_code == 404
def test_private_page_admin_override(client):
owner = _insert_user("owner", "[email protected]", "Owner")
member = _insert_user("member", "[email protected]", "Member")
_, page, _, _ = _scope(owner, member)
_set_page_type(page, "private")
_as(client, member, "member")
assert client.get(f"/board/api/pages/{page}").status_code == 404
# A global admin can always access.
_as(client, member, "member", is_admin=1)
assert client.get(f"/board/api/pages/{page}").status_code == 200
def test_mine_endpoint_and_batch_grant(client):
owner = _insert_user("owner", "[email protected]", "Owner")
member = _insert_user("member", "[email protected]", "Member")
_, page, _, _ = _scope(owner, member)
_set_page_type(page, "restricted")
_as(client, owner, "owner")
mine = client.get(f"/api/v2/pages/{page}/permissions/mine")
assert mine.status_code == 200 and mine.json()["role"] == "owner"
batch = client.post(
f"/api/v2/pages/{page}/permissions/batch",
json={"grants": [{"user_id": member, "role": "viewer"}]},
)
assert batch.status_code == 200 and len(batch.json()["granted"]) == 1
_as(client, member, "member")
mine = client.get(f"/api/v2/pages/{page}/permissions/mine")
assert mine.json()["role"] == "viewer"
assert client.get(f"/board/api/pages/{page}").status_code == 200
def test_page_permission_type_via_api(client):
owner = _insert_user("owner", "[email protected]", "Owner")
_, page, _, _ = _scope(owner)
_as(client, owner, "owner")
resp = client.post(
f"/api/v2/pages/{page}/permission-type",
json={"permission_type": "restricted"},
)
assert resp.status_code == 200
assert resp.json()["permission_type"] == "restricted"
# Listing reflects the type.
listing = client.get(f"/api/v2/pages/{page}/permissions").json()
assert listing["permission_type"] == "restricted"
# ═══════════════ collection permissions ═══════════════
def test_restricted_collection_hidden_for_member(client):
owner = _insert_user("owner", "[email protected]", "Owner")
member = _insert_user("member", "[email protected]", "Member")
ws = _insert_workspace(owner)
coll = _insert_collection(ws)
_set_collection_type(coll, "restricted")
from app.db import get_conn
with get_conn() as conn:
conn.execute(
"INSERT INTO workspace_members (workspace_id, user_id, role) VALUES (?,?,'viewer')",
(ws, member),
)
conn.commit()
_as(client, member, "member")
assert client.get(f"/db/{coll}/api").status_code == 404
# HTML pages 404→redirect to /workspaces (main.py `not_found_handler`);
# a restricted collection is therefore hidden, not served.
assert client.get(f"/db/{coll}", follow_redirects=False).status_code == 302
# Owner sees it.
_as(client, owner, "owner")
assert client.get(f"/db/{coll}/api").status_code == 200
def test_collection_grant_allows_view_then_edit_gate(client):
owner = _insert_user("owner", "[email protected]", "Owner")
member = _insert_user("member", "[email protected]", "Member")
ws = _insert_workspace(owner)
coll = _insert_collection(ws)
_set_collection_type(coll, "restricted")
from app.db import get_conn
with get_conn() as conn:
conn.execute("INSERT INTO workspace_members (workspace_id, user_id, role) VALUES (?,?,'viewer')",
(ws, member))
conn.commit()
_as(client, member, "member")
# Without grant → 404.
assert client.post(f"/db/{coll}/pages/api", json={"title": "New"}).status_code == 404
_as(client, owner, "owner")
grant = client.post(
f"/api/v2/collections/{coll}/permissions",
json={"user_id": member, "role": "viewer"},
)
assert grant.status_code == 200, grant.text
_as(client, member, "member")
# Viewer can read the collection…
assert client.get(f"/db/{coll}/api").status_code == 200
# …but cannot create pages.
assert client.post(f"/db/{coll}/pages/api", json={"title": "New"}).status_code == 403
# …and cannot delete the collection.
assert client.delete(f"/db/api/{coll}").status_code == 403
def test_owner_can_delete_collection(client):
owner = _insert_user("owner", "[email protected]", "Owner")
ws = _insert_workspace(owner)
coll = _insert_collection(ws)
_as(client, owner, "owner")
assert client.delete(f"/db/api/{coll}").status_code == 200
def test_collection_editor_can_create_pages(client):
owner = _insert_user("owner", "[email protected]", "Owner")
member = _insert_user("member", "[email protected]", "Member")
ws = _insert_workspace(owner)
coll = _insert_collection(ws)
from app.db import get_conn
with get_conn() as conn:
conn.execute("INSERT INTO workspace_members (workspace_id, user_id, role) VALUES (?,?,'editor')",
(ws, member))
conn.commit()
_as(client, member, "member")
resp = client.post(f"/db/{coll}/pages/api", json={"title": "New"})
assert resp.status_code == 200, resp.text
# ═══════════════ property visibility ═══════════════
def test_property_permissions_visibility(client):
owner = _insert_user("owner", "[email protected]", "Owner")
other = _insert_user("other", "[email protected]", "Other")
ws = _insert_workspace(owner)
coll = _insert_collection(ws)
p_open = _insert_property(coll, "Open")
p_secret = _insert_property(coll, "Secret")
from app.db import get_conn
with get_conn() as conn:
conn.execute("INSERT INTO workspace_members (workspace_id, user_id, role) VALUES (?,?,'viewer')",
(ws, other))
conn.commit()
# Owner grants view on the Secret property to `other`.
_as(client, owner, "owner")
grant = client.post(
f"/api/v2/collections/{coll}/properties/{p_secret}/permissions",
json={"user_id": other, "role": "viewer"},
)
assert grant.status_code == 200, grant.text
_as(client, other, "other")
props = client.get(f"/db/{coll}/properties/api").json()["properties"]
ids = {p["id"] for p in props}
# Open property (no grants) visible; Secret visible because granted.
assert p_open in ids and p_secret in ids
# Can view but not edit the Secret property.
visible = client.get(f"/api/v2/collections/{coll}/properties/visible").json()
assert p_secret in visible["visible"]
def test_property_hidden_for_non_grantee(client):
owner = _insert_user("owner", "[email protected]", "Owner")
snoop = _insert_user("snoop", "[email protected]", "Snoop")
other = _insert_user("other", "[email protected]", "Other")
ws = _insert_workspace(owner)
coll = _insert_collection(ws)
p_secret = _insert_property(coll, "Secret")
from app.db import get_conn
with get_conn() as conn:
for uid, role in ((snoop, "viewer"), (other, "viewer")):
conn.execute("INSERT INTO workspace_members (workspace_id, user_id, role) VALUES (?,?,?)",
(ws, uid, role))
conn.commit()
# Only `other` is granted — `snoop` has no grant.
_as(client, owner, "owner")
client.post(
f"/api/v2/collections/{coll}/properties/{p_secret}/permissions",
json={"user_id": other, "role": "viewer"},
)
_as(client, snoop, "snoop")
props = client.get(f"/db/{coll}/properties/api").json()["properties"]
assert all(p["id"] != p_secret for p in props)
visible = client.get(f"/api/v2/collections/{coll}/properties/visible").json()
assert p_secret in visible["hidden"]
# Owner still sees everything.
_as(client, owner, "owner")
props = client.get(f"/db/{coll}/properties/api").json()["properties"]
assert {p["id"] for p in props} == {p_secret}
def test_property_grant_requires_collection_owner(client):
owner = _insert_user("owner", "[email protected]", "Owner")
member = _insert_user("member", "[email protected]", "Member")
ws = _insert_workspace(owner)
coll = _insert_collection(ws)
p = _insert_property(coll)
from app.db import get_conn
with get_conn() as conn:
conn.execute("INSERT INTO workspace_members (workspace_id, user_id, role) VALUES (?,?,'editor')",
(ws, member))
conn.commit()
_as(client, member, "member")
resp = client.post(
f"/api/v2/collections/{coll}/properties/{p}/permissions",
json={"user_id": member, "role": "viewer"},
)
assert resp.status_code == 403
# ═══════════════ groups ═══════════════
def test_group_grant_inherits_to_members(client):
owner = _insert_user("owner", "[email protected]", "Owner")
member = _insert_user("member", "[email protected]", "Member")
ws, page, _, _ = _scope(owner, member)
_set_page_type(page, "restricted")
_as(client, owner, "owner")
gid = client.post(
"/api/v2/groups", json={"name": "Editors", "workspace_id": ws}
).json()["id"]
assert client.post(f"/api/v2/groups/{gid}/members", json={"user_id": member}).status_code == 200
assert client.post(
f"/api/v2/pages/{page}/permissions",
json={"group_id": gid, "role": "viewer"},
).status_code == 200
_as(client, member, "member")
assert client.get(f"/board/api/pages/{page}").status_code == 200
# Listing shows the group grant on the page.
_as(client, owner, "owner")
grants = client.get(f"/api/v2/pages/{page}/permissions").json()["permissions"]
assert any(g["kind"] == "group" and g["name"] == "Editors" for g in grants)
def test_group_member_removal_revokes_group_access(client):
owner = _insert_user("owner", "[email protected]", "Owner")
member = _insert_user("member", "[email protected]", "Member")
ws, page, _, _ = _scope(owner, member)
_set_page_type(page, "restricted")
_as(client, owner, "owner")
gid = client.post(
"/api/v2/groups", json={"name": "G", "workspace_id": ws}
).json()["id"]
client.post(f"/api/v2/groups/{gid}/members", json={"user_id": member})
client.post(f"/api/v2/pages/{page}/permissions", json={"group_id": gid, "role": "viewer"})
_as(client, member, "member")
assert client.get(f"/board/api/pages/{page}").status_code == 200
_as(client, owner, "owner")
assert client.delete(f"/api/v2/groups/{gid}/members/{member}").status_code == 200
_as(client, member, "member")
assert client.get(f"/board/api/pages/{page}").status_code == 404
def test_groups_listing_and_members(client):
owner = _insert_user("owner", "[email protected]", "Owner")
member = _insert_user("member", "[email protected]", "Member")
ws = _insert_workspace(owner)
_as(client, owner, "owner")
gid = client.post(
"/api/v2/groups", json={"name": "Team", "workspace_id": ws}
).json()["id"]
client.post(f"/api/v2/groups/{gid}/members", json={"user_id": member})
groups = client.get(f"/api/v2/groups?workspace_id={ws}").json()["groups"]
assert len(groups) == 1 and groups[0]["member_count"] == 1
members = client.get(f"/api/v2/groups/{gid}/members").json()["members"]
assert members[0]["id"] == member
# Non-owner cannot add members.
_as(client, member, "member")
resp = client.post(f"/api/v2/groups/{gid}/members", json={"user_id": owner})
assert resp.status_code == 403
# ═══════════════ audit log + auth ═══════════════
def test_audit_log_records_changes(client):
owner = _insert_user("owner", "[email protected]", "Owner")
member = _insert_user("member", "[email protected]", "Member")
_, page, _, _ = _scope(owner, member)
_as(client, owner, "owner")
client.post(f"/api/v2/pages/{page}/permissions", json={"user_id": member, "role": "viewer"})
client.post(f"/api/v2/pages/{page}/permission-type", json={"permission_type": "restricted"})
events = client.get("/api/v2/audit/permissions").json()["events"]
actions = [e["action"] for e in events]
assert "grant" in actions and "type_change" in actions
# Only the owner workspace can view the log.
_as(client, member, "member")
assert client.get("/api/v2/audit/permissions").status_code == 403
def test_permissions_endpoints_require_auth(client):
assert client.get("/api/v2/pages/1/permissions").status_code == 401
assert client.post("/api/v2/pages/1/permissions", json={"user_id": 2, "role": "viewer"}).status_code == 401
assert client.get("/api/v2/groups").status_code == 401
assert client.get("/api/v2/audit/permissions").status_code == 401
def test_grant_requires_valid_user_or_group(client):
owner = _insert_user("owner", "[email protected]", "Owner")
_, page, _, _ = _scope(owner)
_as(client, owner, "owner")
# Unknown target user.
resp = client.post(f"/api/v2/pages/{page}/permissions", json={"user_id": 99999, "role": "viewer"})
assert resp.status_code == 404
# Unknown group.
resp = client.post(f"/api/v2/pages/{page}/permissions", json={"group_id": 99999, "role": "viewer"})
assert resp.status_code == 404
# Invalid role.
resp = client.post(f"/api/v2/pages/{page}/permissions", json={"user_id": owner, "role": "boss"})
assert resp.status_code == 400