tegwick
1139a6d30e
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a192e9-c28b-7533-bf25-e2341364c22a
239 lines
12 KiB
Python
239 lines
12 KiB
Python
"""Author-managed attachments preserve permissions, revisions and crash recovery."""
|
|
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
import pytest
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from django.test import Client
|
|
from django.urls import reverse
|
|
|
|
from coulomb_social.apps.spaces import attachments, pageops
|
|
from coulomb_social.apps.spaces.assets import local_download_url
|
|
from coulomb_social.apps.spaces.content import load_space_page
|
|
from coulomb_social.apps.spaces.models import SpaceMembership
|
|
from coulomb_social.apps.spaces.services import create_space, put_space_page
|
|
from tests.test_space_permissions import make_member
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
@pytest.fixture
|
|
def people(tmp_path, settings, client):
|
|
settings.CONTENT_ROOT = str(tmp_path / "content")
|
|
owner, reader, outsider = [make_member(name) for name in ("author", "reader", "outsider")]
|
|
space = create_space(owner, title="Shared files", slug="shared-files")
|
|
SpaceMembership.objects.create(space=space, member=reader, role="viewer")
|
|
doc = put_space_page(owner, space, "notes", title="Notes", body="Retained notes",
|
|
extra_meta={"source_created_at": "2020-01-02T03:04:05Z"})
|
|
client.force_login(owner.user)
|
|
return space, doc, owner, reader, outsider
|
|
|
|
|
|
def upload(space, doc, name="Report.pdf", data=b"synthetic PDF"):
|
|
return attachments.add(space.slug, doc.slug, SimpleUploadedFile(name, data),
|
|
revision=doc.revision)
|
|
|
|
|
|
def test_upload_download_and_remove_preserve_source_and_retained_revisions(people, client):
|
|
space, doc, _, reader, _ = people
|
|
url = reverse("spaces:page_attachments", args=[space.slug, doc.slug])
|
|
result = client.post(url, {"action": "add", "revision": doc.revision,
|
|
"file": SimpleUploadedFile("Report.pdf", b"synthetic PDF")})
|
|
assert result.status_code == 302
|
|
added = pageops.read_page(space.slug, doc.slug)
|
|
entry, = attachments.entries(added.meta)
|
|
assert entry["name"] == "Report.pdf"
|
|
assert added.body == doc.body
|
|
assert added.meta["source_created_at"] == doc.meta["source_created_at"]
|
|
client.force_login(reader.user)
|
|
response = client.get(local_download_url(space.slug, entry["path"]))
|
|
assert response.status_code == 200
|
|
assert response["Content-Type"] == "application/octet-stream"
|
|
assert response["Content-Disposition"].startswith("attachment;")
|
|
assert response["Cache-Control"] == "private, no-store"
|
|
assert response["X-Content-Type-Options"] == "nosniff"
|
|
assert b"".join(response.streaming_content) == b"synthetic PDF"
|
|
client.force_login(people[2].user)
|
|
assert client.post(url, {"action": "remove", "revision": added.revision,
|
|
"reference": entry["path"]}).status_code == 302
|
|
current = pageops.read_page(space.slug, doc.slug)
|
|
assert attachments.entries(current.meta) == [] and current.body == doc.body
|
|
assert (pageops.space_dir(space.slug) / entry["path"]).read_bytes() == b"synthetic PDF"
|
|
retained = pageops.content_root() / ".revisions" / space.slug / doc.slug
|
|
assert (retained / (doc.revision + ".md")).is_file()
|
|
assert entry["path"] in (retained / (added.revision + ".md")).read_text()
|
|
|
|
|
|
def test_attachment_permissions_and_revocation_cover_reads_and_mutations(people, client):
|
|
space, doc, owner, reader, outsider = people
|
|
added = upload(space, doc)
|
|
entry, = attachments.entries(added.meta)
|
|
url = reverse("spaces:page_attachments", args=[space.slug, doc.slug])
|
|
download = local_download_url(space.slug, entry["path"])
|
|
foreign = make_member("foreign", tenant="tenant:b")
|
|
SpaceMembership.objects.create(space=space, member=foreign, role="owner")
|
|
client.force_login(reader.user)
|
|
assert client.get(url).status_code == 200
|
|
assert b"Add attachment" not in client.get(url).content
|
|
assert client.post(url, {"action": "remove"}).status_code == 403
|
|
for member in (outsider, foreign):
|
|
client.force_login(member.user)
|
|
assert client.get(url).status_code == client.get(download).status_code == 404
|
|
assert client.post(url, {"action": "add"}).status_code == 404
|
|
SpaceMembership.objects.filter(space=space, member=reader).delete()
|
|
client.force_login(reader.user)
|
|
assert client.get(url).status_code == client.get(download).status_code == 404
|
|
client.force_login(owner.user)
|
|
assert client.post(download).status_code == 405
|
|
space.is_active = False
|
|
space.save()
|
|
assert client.get(download).status_code == 404
|
|
client.logout()
|
|
assert client.get(url).status_code == client.get(download).status_code == 302
|
|
|
|
|
|
def test_stale_missing_or_foreign_attachment_actions_do_not_change_files(people, client):
|
|
space, doc, owner, _, _ = people
|
|
added = upload(space, doc)
|
|
before = {str(p): p.read_bytes() for p in pageops.content_root().rglob('*') if p.is_file()}
|
|
for revision in (doc.revision, None, ""):
|
|
with pytest.raises(pageops.PageOpsError, match="changed"):
|
|
attachments.add(space.slug, doc.slug, SimpleUploadedFile("more.txt", b"more"),
|
|
revision=revision)
|
|
with pytest.raises(pageops.PageOpsError, match="not found"):
|
|
attachments.remove(space.slug, doc.slug, "assets/elsewhere.txt", revision=added.revision)
|
|
after = {str(p): p.read_bytes() for p in pageops.content_root().rglob('*') if p.is_file()}
|
|
assert after == before
|
|
guarded = Client(enforce_csrf_checks=True)
|
|
guarded.force_login(owner.user)
|
|
url = reverse("spaces:page_attachments", args=[space.slug, doc.slug])
|
|
assert guarded.post(url, {"action": "remove"}).status_code == 403
|
|
|
|
|
|
@pytest.mark.parametrize("path", ["../pages/notes.md", "nested/../../private", "..%2fsecret",
|
|
"missing", "link.txt", "folder/link.txt"])
|
|
def test_download_refuses_missing_traversal_and_symlinks(people, client, tmp_path, path):
|
|
space, _, _, _, _ = people
|
|
secret = tmp_path / "secret.txt"
|
|
secret.write_text("private")
|
|
(pageops.assets_dir(space.slug) / "link.txt").symlink_to(secret)
|
|
(pageops.assets_dir(space.slug) / "folder").symlink_to(tmp_path, target_is_directory=True)
|
|
assert client.get(reverse("spaces:download", args=[space.slug, path])).status_code == 404
|
|
|
|
|
|
def test_active_content_is_download_only_and_filenames_are_escaped(people, client):
|
|
space, doc, _, _, _ = people
|
|
added = upload(space, doc, '<script>alert(1)</script>.html', b'<script>synthetic()</script>')
|
|
entry, = attachments.entries(added.meta)
|
|
response = client.get(local_download_url(space.slug, entry["path"]))
|
|
assert response["Content-Disposition"].startswith("attachment;")
|
|
assert response["Content-Type"] == "application/octet-stream"
|
|
assert "sandbox" in response["Content-Security-Policy"]
|
|
assert b"".join(response.streaming_content) == b'<script>synthetic()</script>'
|
|
menu = client.get(reverse("spaces:page_attachments", args=[space.slug, doc.slug]))
|
|
assert b"<script>" not in menu.content
|
|
|
|
|
|
def test_inline_body_files_have_authenticated_download_links(people, client):
|
|
space, doc, _, _, _ = people
|
|
(pageops.assets_dir(space.slug) / "legacy.txt").write_bytes(b"legacy bytes")
|
|
doc = pageops.write_page(space.slug, doc.slug, title="Notes",
|
|
body="[Existing resource](assets/legacy.txt)")
|
|
file, = attachments.page_files(space.slug, doc)
|
|
assert not file["managed"]
|
|
assert file["download_url"] in load_space_page(space, doc.slug).html
|
|
menu = client.get(reverse("spaces:page_attachments", args=[space.slug, doc.slug]))
|
|
assert b"Edit page reference" in menu.content
|
|
|
|
|
|
def test_edit_and_rename_keep_attachment_associations(people, client):
|
|
space, doc, _, _, _ = people
|
|
added = upload(space, doc)
|
|
response = client.post(reverse("spaces:page_edit", args=[space.slug, doc.slug]), {
|
|
"title": "Renamed notes", "slug": "renamed", "body": "Updated notes",
|
|
"revision": added.revision,
|
|
})
|
|
assert response.status_code == 302
|
|
changed = pageops.read_page(space.slug, "renamed")
|
|
assert changed.body == "Updated notes\n"
|
|
assert attachments.entries(changed.meta) == attachments.entries(added.meta)
|
|
assert changed.meta["source_created_at"] == doc.meta["source_created_at"]
|
|
|
|
|
|
@pytest.mark.parametrize("operation", ["copy_page", "transfer_page"])
|
|
def test_copy_move_and_detach_keep_attachment_bytes_independent(people, operation):
|
|
space, doc, owner, _, _ = people
|
|
added = upload(space, doc)
|
|
target = create_space(owner, title="Other", slug="other")
|
|
copied = getattr(pageops, operation)(space.slug, doc.slug, target.slug, "copied")
|
|
original, = attachments.entries(added.meta)
|
|
entry, = attachments.entries(copied.meta)
|
|
assert entry["path"] != original["path"]
|
|
assert (pageops.space_dir(target.slug) / entry["path"]).read_bytes() == b"synthetic PDF"
|
|
attachments.remove(target.slug, copied.slug, entry["path"], revision=copied.revision)
|
|
assert (pageops.space_dir(space.slug) / original["path"]).read_bytes() == b"synthetic PDF"
|
|
assert pageops.recover_interrupted_content() == {"cleared": 0, "held": 0}
|
|
|
|
|
|
@pytest.mark.parametrize("phase", ["partial", "bytes_saved", "revision_saved", "page_saved"])
|
|
def test_upload_process_death_preserves_page_or_complete_retained_attachment(people, phase):
|
|
space, doc, _, _, _ = people
|
|
code = r'''
|
|
import os,sys
|
|
from django.conf import settings
|
|
settings.configure(CONTENT_ROOT=sys.argv[1])
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from coulomb_social.apps.spaces import attachments,pageops
|
|
phase=sys.argv[4]
|
|
opening=attachments.open_local_asset
|
|
def open_asset(*a,**kw):
|
|
f=opening(*a,**kw)
|
|
if phase=='partial' and kw.get('create'):
|
|
f.write(b'partial'); f.flush(); os.fsync(f.fileno()); os._exit(86)
|
|
return f
|
|
attachments.open_local_asset=open_asset
|
|
save=attachments._save
|
|
def saving(*a):
|
|
if phase=='bytes_saved':os._exit(86)
|
|
return save(*a)
|
|
attachments._save=saving
|
|
remember=pageops._remember
|
|
def remembering(space,page,data):
|
|
remember(space,page,data)
|
|
if phase=='revision_saved' and b'coulomb_attachments' in data:os._exit(86)
|
|
pageops._remember=remembering
|
|
clear=pageops._clear_record
|
|
def clearing(*a):
|
|
if phase=='page_saved':os._exit(86)
|
|
return clear(*a)
|
|
pageops._clear_record=clearing
|
|
attachments.add(sys.argv[2],'notes',SimpleUploadedFile('report.txt',b'complete bytes'),
|
|
revision=sys.argv[3])
|
|
'''
|
|
result = subprocess.run([sys.executable, "-c", code, str(pageops.content_root()),
|
|
space.slug, doc.revision, phase], capture_output=True, timeout=20,
|
|
env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"})
|
|
assert result.returncode == 86, result.stderr.decode()
|
|
assert pageops.recover_interrupted_content() == {"cleared": 1, "held": 0}
|
|
current = pageops.read_page(space.slug, doc.slug)
|
|
assert current.body == doc.body
|
|
assets = list(pageops.assets_dir(space.slug).iterdir())
|
|
if phase in {"partial", "bytes_saved"}:
|
|
assert current.revision == doc.revision and not assets
|
|
else:
|
|
assert len(assets) == 1 and assets[0].read_bytes() == b"complete bytes"
|
|
retained = pageops.content_root() / ".revisions" / space.slug / doc.slug
|
|
assert any(attachments.KEY in p.read_text() for p in retained.glob("*.md"))
|
|
if phase == "page_saved":
|
|
assert len(attachments.entries(current.meta)) == 1
|
|
|
|
|
|
def test_oversized_upload_refuses_before_file_or_page_write(people, monkeypatch):
|
|
space, doc, _, _, _ = people
|
|
monkeypatch.setattr(attachments, "MAX_UPLOAD_BYTES", 4)
|
|
with pytest.raises(pageops.PageOpsError, match="16 MiB"):
|
|
upload(space, doc, data=b"too large")
|
|
assert pageops.read_page(space.slug, doc.slug).revision == doc.revision
|
|
assert not list(pageops.assets_dir(space.slug).iterdir())
|