feat: restrict backup archive download to primary admin and hide admin-hidden projects from other admins

- Add `get_admin_uids()` and `get_primary_admin_uid()` to database.py for resolving the earliest-created admin
- Modify `can_view_project()` in content.py so a project hidden by an admin is invisible to other admins (both web UI and REST API)
- Update `_download_url()` and `_backup_payload()` in admin/backups.py to accept a `can_download` flag, gating the download endpoint with `is_primary_admin()`
- Remove `role` from `_user_facts()` in docs_live.py to avoid leaking admin status in live docs
- Update doc summaries in docs_api.py to reflect the new admin-visibility and backup-download semantics
This commit is contained in:
2026-06-17 14:08:28 +00:00
parent 6b5347103b
commit 0a554ebc32
71 changed files with 1868 additions and 527 deletions
View File
+129
View File
@@ -0,0 +1,129 @@
# retoor <retoor@molodetz.nl>
import time
import requests
from tests.conftest import BASE_URL
from devplacepy import config
from devplacepy.database import get_table, get_primary_admin_uid, refresh_snapshot
from devplacepy.services.backup import store
from devplacepy.utils import clear_user_cache
JSON = {"Accept": "application/json"}
_counter = [0]
def _signup():
_counter[0] += 1
name = f"bkp{int(time.time() * 1000)}{_counter[0]}"
requests.post(
f"{BASE_URL}/auth/signup",
data={
"username": name,
"email": f"{name}@t.dev",
"password": "secret123",
"confirm_password": "secret123",
},
allow_redirects=True,
)
refresh_snapshot()
return get_table("users").find_one(username=name)
def _make_admin():
row = _signup()
get_table("users").update({"uid": row["uid"], "role": "Admin"}, ["uid"])
clear_user_cache(row["uid"])
return row["api_key"]
def _primary_admin_key():
refresh_snapshot()
uid = get_primary_admin_uid()
return get_table("users").find_one(uid=uid)["api_key"]
def _seed_done_backup(owner_uid):
backup_uid = store.create_backup(
target="database", created_by=owner_uid, job_uid="jobtest"
)
target_dir = config.BACKUPS_DIR
target_dir.mkdir(parents=True, exist_ok=True)
path = target_dir / f"{backup_uid}.tar.gz"
path.write_bytes(b"\x1f\x8b\x08\x00test-archive")
store.finalize_backup(
backup_uid,
filename="database-test.tar.gz",
local_path=str(path),
stats={"bytes_out": path.stat().st_size, "file_count": 1},
)
refresh_snapshot()
return backup_uid
def test_download_requires_admin_guest(app_server):
assert (
requests.get(
f"{BASE_URL}/admin/backups/nope/download",
headers=JSON,
allow_redirects=False,
).status_code
== 401
)
assert (
requests.get(
f"{BASE_URL}/admin/backups/nope/download", allow_redirects=False
).status_code
== 303
)
def test_download_denied_for_member(app_server, seeded_db):
s = requests.Session()
s.post(
f"{BASE_URL}/auth/login",
data={
"email": seeded_db["bob"]["email"],
"password": seeded_db["bob"]["password"],
},
)
r = s.get(
f"{BASE_URL}/admin/backups/nope/download", headers=JSON, allow_redirects=False
)
assert r.status_code == 403
def test_download_and_flag_denied_for_non_primary_admin(app_server):
key = _make_admin()
headers = {**JSON, "X-API-KEY": key}
assert (
requests.get(
f"{BASE_URL}/admin/backups/nope/download",
headers=headers,
allow_redirects=False,
).status_code
== 403
)
data = requests.get(f"{BASE_URL}/admin/backups/data", headers=headers).json()
assert data["can_download_backups"] is False
assert all(b.get("download_url") is None for b in data["backups"])
def test_primary_admin_flag_true(app_server):
headers = {**JSON, "X-API-KEY": _primary_admin_key()}
data = requests.get(f"{BASE_URL}/admin/backups/data", headers=headers).json()
assert data["can_download_backups"] is True
def test_primary_admin_can_download_archive(app_server):
uid = get_primary_admin_uid()
key = get_table("users").find_one(uid=uid)["api_key"]
backup_uid = _seed_done_backup(uid)
headers = {**JSON, "X-API-KEY": key}
data = requests.get(f"{BASE_URL}/admin/backups/data", headers=headers).json()
rows = [b for b in data["backups"] if b["uid"] == backup_uid]
assert rows and rows[0]["download_url"] == f"/admin/backups/{backup_uid}/download"
r = requests.get(f"{BASE_URL}/admin/backups/{backup_uid}/download", headers=headers)
assert r.status_code == 200
assert r.headers["content-type"] == "application/gzip"
@@ -1,8 +1,52 @@
# retoor <retoor@molodetz.nl>
import time
import requests
from tests.conftest import BASE_URL
from devplacepy.database import get_table, refresh_snapshot
from devplacepy.utils import clear_user_cache
_counter_search = [0]
def _signup_search():
_counter_search[0] += 1
name = f"cas{int(time.time() * 1000)}{_counter_search[0]}"
requests.post(
f"{BASE_URL}/auth/signup",
data={
"username": name,
"email": f"{name}@t.dev",
"password": "secret123",
"confirm_password": "secret123",
},
allow_redirects=True,
)
row = get_table("users").find_one(username=name)
return name, row["uid"], row["api_key"]
def _make_admin_search():
name, uid, key = _signup_search()
get_table("users").update({"uid": uid, "role": "Admin"}, ["uid"])
clear_user_cache(uid)
return name, uid, key
def _create_private_project(key, title):
r = requests.post(
f"{BASE_URL}/projects/create",
headers={"X-API-KEY": key, "Accept": "application/json"},
data={
"title": title,
"description": "search test",
"project_type": "software",
"status": "In Development",
"is_private": "on",
},
)
assert r.status_code == 200, r.text
return r.json()["data"]["slug"]
def _admin_session():
@@ -33,3 +77,21 @@ def test_project_search_empty_query(app_server, seeded_db):
r = s.get(f"{BASE_URL}/admin/containers/projects/search?q=")
assert r.status_code == 200
assert r.json()["results"] == []
def test_project_search_hides_other_admin_private_project(app_server):
_, _, owner_key = _make_admin_search()
_, _, other_admin_key = _make_admin_search()
title = f"SearchHidden{int(time.time() * 1000)}"
slug = _create_private_project(owner_key, title)
other = requests.Session()
other.headers.update({"X-API-KEY": other_admin_key})
r = other.get(f"{BASE_URL}/admin/containers/projects/search?q={title}")
assert r.status_code == 200
assert slug not in [row["slug"] for row in r.json()["results"]]
owner = requests.Session()
owner.headers.update({"X-API-KEY": owner_key})
r = owner.get(f"{BASE_URL}/admin/containers/projects/search?q={title}")
assert slug in [row["slug"] for row in r.json()["results"]]
+166
View File
@@ -0,0 +1,166 @@
# retoor <retoor@molodetz.nl>
import time
import pytest
import requests
from tests.conftest import BASE_URL
from devplacepy.database import get_table, refresh_snapshot, set_setting
JSON = {"Accept": "application/json"}
_counter = [0]
@pytest.fixture(scope="module", autouse=True)
def _settings(app_server):
for key, value in {
"rate_limit_per_minute": "1000000",
"rate_limit_window_seconds": "60",
"registration_open": "1",
"maintenance_mode": "0",
"session_max_age_days": "7",
"session_remember_days": "30",
}.items():
set_setting(key, value)
yield
def _db_user(name):
refresh_snapshot()
return get_table("users").find_one(username=name)
def _unique(prefix="sen"):
_counter[0] += 1
return f"{prefix}{int(time.time() * 1000)}{_counter[0]}"
def _signup():
name = _unique("senmem")
s = requests.Session()
s.post(
f"{BASE_URL}/auth/signup",
data={
"username": name,
"email": f"{name}@t.dev",
"password": "secret123",
"confirm_password": "secret123",
},
allow_redirects=True,
)
return name
def _key_session(name):
s = requests.Session()
s.headers.update({"X-API-KEY": _db_user(name)["api_key"]})
return s
def _alice():
return _key_session("alice_test")
def _promote(actor, uid):
actor.post(
f"{BASE_URL}/admin/users/{uid}/role",
headers=JSON,
data={"role": "admin"},
allow_redirects=False,
)
def _audit(session, **params):
r = session.get(f"{BASE_URL}/admin/audit-log", headers=JSON, params=params)
assert r.status_code == 200, r.text[:300]
return r.json()["entries"]
def _denied(session, event_key, target_uid):
for e in _audit(session, event_key=event_key, result="denied"):
if e.get("target_uid") == target_uid:
return e
return None
def _junior_admin(seeded_db):
alice = _alice()
name = _signup()
uid = _db_user(name)["uid"]
_promote(alice, uid)
assert _db_user(name)["role"] == "Admin"
return _key_session(name), name, uid
def test_junior_admin_cannot_change_senior_admin_role(seeded_db):
junior, _, _ = _junior_admin(seeded_db)
alice_uid = _db_user("alice_test")["uid"]
junior.post(
f"{BASE_URL}/admin/users/{alice_uid}/role",
headers=JSON,
data={"role": "member"},
allow_redirects=False,
)
assert _denied(_alice(), "admin.user.role.change", alice_uid) is not None
assert _db_user("alice_test")["role"] == "Admin"
def test_junior_admin_cannot_disable_senior_admin(seeded_db):
junior, _, _ = _junior_admin(seeded_db)
alice_uid = _db_user("alice_test")["uid"]
junior.post(
f"{BASE_URL}/admin/users/{alice_uid}/toggle",
headers=JSON,
allow_redirects=False,
)
assert _denied(_alice(), "admin.user.active.disable", alice_uid) is not None
assert _db_user("alice_test").get("is_active", True)
def test_junior_admin_cannot_reset_senior_admin_password(seeded_db):
junior, _, _ = _junior_admin(seeded_db)
alice_uid = _db_user("alice_test")["uid"]
before = _db_user("alice_test")["password_hash"]
junior.post(
f"{BASE_URL}/admin/users/{alice_uid}/password",
headers=JSON,
data={"password": "hijacked999"},
allow_redirects=False,
)
assert _denied(_alice(), "admin.user.password.reset", alice_uid) is not None
assert _db_user("alice_test")["password_hash"] == before
def test_junior_admin_cannot_reset_senior_admin_quota(seeded_db):
junior, _, _ = _junior_admin(seeded_db)
alice_uid = _db_user("alice_test")["uid"]
junior.post(
f"{BASE_URL}/admin/users/{alice_uid}/reset-ai-quota",
headers=JSON,
allow_redirects=False,
)
assert _denied(_alice(), "admin.user.ai_quota.reset", alice_uid) is not None
def test_senior_admin_can_disable_junior_admin(seeded_db):
_junior, _, junior_uid = _junior_admin(seeded_db)
alice = _alice()
alice.post(
f"{BASE_URL}/admin/users/{junior_uid}/toggle",
headers=JSON,
allow_redirects=False,
)
refresh_snapshot()
assert not get_table("users").find_one(uid=junior_uid).get("is_active", True)
def test_junior_admin_can_manage_member(seeded_db):
junior, _, _ = _junior_admin(seeded_db)
member = _signup()
member_uid = _db_user(member)["uid"]
junior.post(
f"{BASE_URL}/admin/users/{member_uid}/toggle",
headers=JSON,
allow_redirects=False,
)
refresh_snapshot()
assert not get_table("users").find_one(uid=member_uid).get("is_active", True)
+38 -35
View File
@@ -1,51 +1,54 @@
# retoor <retoor@molodetz.nl>
MARK = "_apitest_dbapi_crud"
from devplacepy.database import get_table, purge
from devplacepy.utils import generate_uid
MARK = "_apitest_dbapi_readonly"
def test_list_requires_auth(client):
assert client.get("/dbapi/users").status_code == 403
def test_crud_round_trip(client, auth):
created = client.post(
def test_reads_work(client, auth):
uid = generate_uid()
get_table("bookmarks").insert(
{
"uid": uid,
"user_uid": MARK,
"target_type": "post",
"target_uid": "t1",
"deleted_at": None,
"deleted_by": None,
}
)
try:
listing = client.get(f"/dbapi/bookmarks?filter.user_uid={MARK}", headers=auth)
assert listing.status_code == 200
assert listing.json()["count"] == 1
got = client.get(f"/dbapi/bookmarks/uid/{uid}", headers=auth)
assert got.status_code == 200
assert got.json()["row"]["uid"] == uid
finally:
purge("bookmarks", user_uid=MARK)
def test_writes_are_not_available(client, auth):
insert = client.post(
"/dbapi/bookmarks",
json={"user_uid": MARK, "target_type": "post", "target_uid": "t1"},
headers=auth,
)
assert created.status_code == 200
row = created.json()["row"]
uid = row["uid"]
assert row["deleted_at"] is None
try:
got = client.get(f"/dbapi/bookmarks/uid/{uid}", headers=auth)
assert got.json()["row"]["uid"] == uid
updated = client.patch(
f"/dbapi/bookmarks/uid/{uid}", json={"target_uid": "t2"}, headers=auth
)
assert updated.json()["row"]["target_uid"] == "t2"
deleted = client.delete(f"/dbapi/bookmarks/uid/{uid}", headers=auth)
assert deleted.json()["mode"] == "soft"
live = client.get(f"/dbapi/bookmarks?filter.user_uid={MARK}", headers=auth)
assert live.json()["count"] == 0
with_deleted = client.get(
f"/dbapi/bookmarks?filter.user_uid={MARK}&include_deleted=true", headers=auth
)
assert with_deleted.json()["count"] == 1
restored = client.post(f"/dbapi/bookmarks/uid/{uid}/restore", headers=auth)
assert restored.json()["row"]["deleted_at"] is None
finally:
client.delete(f"/dbapi/bookmarks/uid/{uid}?hard=true", headers=auth)
def test_insert_unknown_column_is_400(client, auth):
response = client.post(
"/dbapi/bookmarks", json={"user_uid": MARK, "no_such_col": 1}, headers=auth
assert insert.status_code in (404, 405)
patched = client.patch(
"/dbapi/bookmarks/uid/whatever", json={"target_uid": "t2"}, headers=auth
)
assert response.status_code == 400
assert patched.status_code in (404, 405)
removed = client.delete("/dbapi/bookmarks/uid/whatever", headers=auth)
assert removed.status_code in (404, 405)
restored = client.post("/dbapi/bookmarks/uid/whatever/restore", headers=auth)
assert restored.status_code in (404, 405)
def test_get_missing_row_is_404(client, auth):
+33
View File
@@ -78,6 +78,16 @@ def _list_slugs(key=None, user_uid=None):
params = {"user_uid": user_uid} if user_uid else None
r = requests.get(f"{BASE_URL}/projects", headers=_h_project_visibility(key), params=params)
return [p["slug"] for p in r.json()["projects"]]
def _detail_status(key, slug):
return requests.get(
f"{BASE_URL}/projects/{slug}", headers=_h_project_visibility(key)
).status_code
def _file_raw_status(key, slug, path):
return requests.get(
f"{BASE_URL}/projects/{slug}/files/raw",
headers=_h_project_visibility(key),
params={"path": path},
).status_code
def test_private_project_hidden_from_guest_listing(app_server):
@@ -101,6 +111,29 @@ def test_private_project_visible_to_admin(app_server):
assert slug in _list_slugs(key=admin_key, user_uid=owner_uid)
def test_admin_private_hidden_from_other_admin(app_server):
_, owner_uid, owner_key = _make_admin_project_visibility()
_, _, other_admin_key = _make_admin_project_visibility()
slug = _create_project_project_visibility(
owner_key, "Admin Hidden", is_private=True
)["slug"]
assert slug not in _list_slugs(key=other_admin_key, user_uid=owner_uid)
assert _detail_status(other_admin_key, slug) == 404
assert _file_raw_status(other_admin_key, slug, "missing.txt") == 404
assert slug in _list_slugs(key=owner_key, user_uid=owner_uid)
assert _detail_status(owner_key, slug) == 200
def test_member_private_still_visible_to_admin(app_server):
_, owner_uid, owner_key = _signup_project_visibility()
_, _, admin_key = _make_admin_project_visibility()
slug = _create_project_project_visibility(
owner_key, "Member Hidden", is_private=True
)["slug"]
assert slug in _list_slugs(key=admin_key, user_uid=owner_uid)
assert _detail_status(admin_key, slug) == 200
def test_toggle_private_then_public(app_server):
_, owner_uid, key = _signup_project_visibility()
slug = _create_project_project_visibility(key, "Toggle Privacy")["slug"]
+84
View File
@@ -0,0 +1,84 @@
# retoor <retoor@molodetz.nl>
import time
import requests
from tests.conftest import BASE_URL
from devplacepy.database import get_table
from devplacepy.utils import clear_user_cache
_counter_cv = [0]
def _signup():
_counter_cv[0] += 1
name = f"cav{int(time.time() * 1000)}{_counter_cv[0]}"
requests.post(
f"{BASE_URL}/auth/signup",
data={
"username": name,
"email": f"{name}@t.dev",
"password": "secret123",
"confirm_password": "secret123",
},
allow_redirects=True,
)
row = get_table("users").find_one(username=name)
return name, row["uid"], row["api_key"]
def _make_admin():
name, uid, key = _signup()
get_table("users").update({"uid": uid, "role": "Admin"}, ["uid"])
clear_user_cache(uid)
return name, uid, key
def _headers(key=None):
headers = {"Accept": "application/json"}
if key:
headers["X-API-KEY"] = key
return headers
def _create_project(key, title, is_private=False):
data = {
"title": title,
"description": "container access test",
"project_type": "software",
"status": "In Development",
}
if is_private:
data["is_private"] = "on"
r = requests.post(
f"{BASE_URL}/projects/create", headers=_headers(key), data=data
)
assert r.status_code == 200, r.text
return r.json()["data"]["slug"]
def _containers_data_status(key, slug):
return requests.get(
f"{BASE_URL}/projects/{slug}/containers/data", headers=_headers(key)
).status_code
def test_admin_private_project_containers_hidden_from_other_admin(app_server):
_, _, owner_key = _make_admin()
_, _, other_admin_key = _make_admin()
slug = _create_project(owner_key, "Admin Container Hidden", is_private=True)
assert _containers_data_status(other_admin_key, slug) == 404
assert _containers_data_status(owner_key, slug) == 200
def test_member_private_project_containers_visible_to_admin(app_server):
_, _, owner_key = _signup()
_, _, admin_key = _make_admin()
slug = _create_project(owner_key, "Member Container Hidden", is_private=True)
assert _containers_data_status(admin_key, slug) == 200
def test_public_project_containers_visible_to_any_admin(app_server):
_, _, owner_key = _make_admin()
_, _, other_admin_key = _make_admin()
slug = _create_project(owner_key, "Public Container", is_private=False)
assert _containers_data_status(other_admin_key, slug) == 200