diff --git a/coulomb_social/apps/spaces/assets.py b/coulomb_social/apps/spaces/assets.py index c4d8572..61d77d4 100644 --- a/coulomb_social/apps/spaces/assets.py +++ b/coulomb_social/apps/spaces/assets.py @@ -83,6 +83,13 @@ def open_local_image(space_slug: str, path: str): def local_asset_url(space_slug: str, reference: str) -> str: """Map space-root or page-relative assets to the authenticated app route.""" + path = local_asset_path(reference) + if not path or PurePosixPath(path).suffix.lower() not in _TYPES: + return "" + return reverse("spaces:asset", kwargs={"slug": space_slug, "asset_path": path}) + + +def local_asset_path(reference: str) -> str: try: parsed = urlsplit(reference) except ValueError: @@ -98,11 +105,16 @@ def local_asset_url(space_slug: str, reference: str) -> str: return "" try: asset_parts(path) - if PurePosixPath(path).suffix.lower() not in _TYPES: - return "" except ValueError: return "" - return reverse("spaces:asset", kwargs={"slug": space_slug, "asset_path": path}) + return path + + +def local_download_url(space_slug: str, reference: str) -> str: + path = local_asset_path(reference) + if not path: + return "" + return reverse("spaces:download", kwargs={"slug": space_slug, "asset_path": path}) class _AssetLinks(HTMLParser): @@ -115,7 +127,9 @@ class _AssetLinks(HTMLParser): result = [] for name, value in attrs: if (tag, name) in {("img", "src"), ("a", "href")} and value: - value = local_asset_url(self.space_slug, value) or value + value = (local_asset_url(self.space_slug, value) + or (local_download_url(self.space_slug, value) if tag == "a" else "") + or value) result.append(name if value is None else f'{name}="{escape(value, quote=True)}"') self.output.append("<" + tag + (" " + " ".join(result) if result else "") + ">") diff --git a/coulomb_social/apps/spaces/attachments.py b/coulomb_social/apps/spaces/attachments.py new file mode 100644 index 0000000..8a907c0 --- /dev/null +++ b/coulomb_social/apps/spaces/attachments.py @@ -0,0 +1,196 @@ +"""Page attachments: immutable files, revisioned associations and bounded uploads.""" + +import hashlib +import json +import os +import re +import uuid +from html.parser import HTMLParser +from pathlib import Path + +from django.utils.text import slugify + +from . import pageops +from .assets import ( + asset_parts, + local_asset_path, + local_asset_url, + local_download_url, + open_local_asset, +) + +KEY = "coulomb_attachments" +MAX_UPLOAD_BYTES = 16 * 1024 * 1024 +MAX_ATTACHMENTS = 100 + + +def page_files(space, doc): + """List page associations plus existing body/Visual references, without I/O.""" + from .content import render_markdown + + files = {local_asset_path(e["path"]): {**e, "managed": True} for e in entries(doc.meta)} + + def add_reference(value): + path = local_asset_path(value) + if path and path not in files: + files[path] = {"path": "assets/" + path, "name": Path(path).name, "managed": False} + + class Links(HTMLParser): + def handle_starttag(self, tag, attrs): + for key, value in attrs: + if (tag, key) in {("a", "href"), ("img", "src")} and value: + add_reference(value) + + parser = Links() + parser.feed(render_markdown(doc.body)) + parser.close() + add_reference(doc.visual) + return [{**entry, "download_url": local_download_url(space, entry["path"]), + "open_url": local_asset_url(space, entry["path"])} for entry in files.values()] + + +def entries(meta): + raw = meta.get(KEY, "[]") + try: + result = json.loads(raw) + if not isinstance(result, list) or len(result) > MAX_ATTACHMENTS: + raise ValueError + paths = set() + for entry in result: + if (not isinstance(entry, dict) or set(entry) != {"name", "path"} + or not isinstance(entry["name"], str) or not entry["name"] + or len(entry["name"]) > 200 or not isinstance(entry["path"], str) + or not entry["path"].startswith("assets/")): + raise ValueError + asset_parts(entry["path"][7:]) + if entry["path"] in paths: + raise ValueError + paths.add(entry["path"]) + return result + except (ValueError, TypeError, KeyError) as exc: + raise pageops.PageOpsError("The attachment list needs operator review") from exc + + +def _metadata(space, doc, files): + meta = dict(title=doc.title, abstractor=doc.abstractor, visual=doc.visual, + space=space) + meta.update({k: v for k, v in doc.meta.items() if k not in meta}) + meta[KEY] = json.dumps(files, ensure_ascii=True) + return meta + + +def _current(space, page, revision): + doc = pageops.read_page(space, page) + if not doc.exists: + raise pageops.PageOpsError("Page not found") + if not revision or doc.revision != revision: + raise pageops.PageOpsError("This page changed. Reopen it before changing attachments") + return doc + + +def _save(space, doc, meta): + return pageops.write_page( + space, doc.slug, title=doc.title, abstractor=doc.abstractor, + visual=doc.visual, body=doc.body, extra_meta=meta, expected_revision=doc.revision, + ) + + +@pageops.content_writer +def add(space, page, upload, *, revision): + doc = _current(space, page, revision) + files = entries(doc.meta) + if len(files) >= MAX_ATTACHMENTS: + raise pageops.PageOpsError("This page already has 100 attachments") + if upload is None or upload.size > MAX_UPLOAD_BYTES: + raise pageops.PageOpsError("Choose a file no larger than 16 MiB") + raw = upload.read(MAX_UPLOAD_BYTES + 1) + if not raw or len(raw) > MAX_UPLOAD_BYTES: + raise pageops.PageOpsError("Choose a non-empty file no larger than 16 MiB") + label = Path(upload.name.replace("\\", "/")).name + label = "".join(c for c in label if c.isprintable())[:200] or "attachment" + suffix = Path(label).suffix.lower() + if not re.fullmatch(r"\.[a-z0-9]{1,12}", suffix): + suffix = "" + stem = slugify(Path(label).stem)[:80] or "attachment" + identity = uuid.uuid4().hex + name = f"upload-{identity}-{stem}{suffix}" + path = pageops.assets_dir(space) / name + # Exclusive creation and the shared writer lock preserve every older asset. + if path.exists() or path.is_symlink(): + raise pageops.PageOpsError("Attachment address already exists") + meta = _metadata(space, doc, [*files, {"name": label, "path": "assets/" + name}]) + record = { + "id": identity, "version": 1, "op": "attachment_upload", "phase": "started", + "source_space": space, "dest_space": space, + "source_page": doc.slug, "dest_page": doc.slug, + "source_revision": doc.revision, + "dest_revision": hashlib.sha256( + pageops.dump_frontmatter(meta, doc.body).encode()).hexdigest(), + "assets": [name], "asset_sha256": hashlib.sha256(raw).hexdigest(), + } + pageops._write_record(record) + try: + with open_local_asset(space, name, create=True) as stream: + stream.write(raw) + stream.flush() + os.fsync(stream.fileno()) + pageops._fsync_directory(path.parent) + result = _save(space, doc, meta) + if result.revision != record["dest_revision"]: + raise pageops.PageOpsError("Attachment publication needs operator review") + pageops._clear_record(record) + return result + except BaseException: + pageops._recover_record(record) + raise + + +@pageops.content_writer +def remove(space, page, reference, *, revision): + doc = _current(space, page, revision) + files = entries(doc.meta) + remaining = [entry for entry in files if entry["path"] != reference] + if len(remaining) == len(files): + raise pageops.PageOpsError("Attachment not found on this page") + # File bytes remain available to retained revisions and other page references. + return _save(space, doc, _metadata(space, doc, remaining)) + + +def recover(record): + """Keep revision-referenced complete bytes; discard only this failed upload.""" + try: + identity = record["id"] + if not re.fullmatch(r"[a-f0-9]{32}", identity): + return "held" + space, page = record["source_space"], record["source_page"] + if (record.get("version") != 1 or record.get("phase") != "started" + or record["dest_space"] != space or record["dest_page"] != page + or len(record["assets"]) != 1): + return "held" + name = record["assets"][0] + if not name.startswith(f"upload-{identity}-") or len(asset_parts(name)) != 1: + return "held" + for field in ("source_revision", "dest_revision", "asset_sha256"): + if not re.fullmatch(r"[a-f0-9]{64}", record[field]): + return "held" + current = pageops.page_file(space, page) + if pageops.assets_dir(space).is_symlink(): + return "held" + revision = pageops.content_root() / ".revisions" / space / page / ( + record["dest_revision"] + ".md") + if any(p.is_symlink() for p in (revision, *revision.parents)): + return "held" + if revision.exists(): + if pageops._file_revision(revision) != record["dest_revision"]: + return "held" + with open_local_asset(space, name) as stream: + if hashlib.file_digest(stream, "sha256").hexdigest() != record["asset_sha256"]: + return "held" + elif current.is_file() and pageops._file_revision(current) == record["source_revision"]: + pageops._discard_recorded_assets(space, [name]) + else: + return "held" + pageops._clear_record(record) + return "cleared" + except (KeyError, TypeError, ValueError, OSError, pageops.PageOpsError): + return "held" diff --git a/coulomb_social/apps/spaces/pageops.py b/coulomb_social/apps/spaces/pageops.py index 0ad7b16..cc48cce 100644 --- a/coulomb_social/apps/spaces/pageops.py +++ b/coulomb_social/apps/spaces/pageops.py @@ -448,6 +448,9 @@ def _finish_transfer_record(record, source_path, dest_confirmed): def _recover_record(record): + if isinstance(record, dict) and record.get("op") == "attachment_upload": + from .attachments import recover + return recover(record) if record == "held" or not _record_is_actionable(record): logger.error("content recovery left a journal for operator review") return "held" @@ -661,7 +664,8 @@ _BODY_ASSET = re.compile( ) -def _copy_assets(source_space, dest_space, dest_page, visual, body, before_create=None): +def _copy_assets(source_space, dest_space, dest_page, visual, body, before_create=None, + attached=()): """Copy local Visual and Markdown link/image destinations as one staged set.""" from .assets import asset_parts, open_local_asset @@ -725,7 +729,8 @@ def _copy_assets(source_space, dest_space, dest_page, visual, body, before_creat body = _BODY_ASSET.sub( lambda match: match["prefix"] + copy_reference(match["url"]), body, ) - return visual, body, created + attached = [{**entry, "path": copy_reference(entry["path"])} for entry in attached] + return visual, body, created, attached except BaseException: for path in created: path.unlink(missing_ok=True) @@ -764,8 +769,11 @@ def copy_page( on_asset(name) try: - visual, body, created = _copy_assets( + from .attachments import KEY, entries + + visual, body, created, attached = _copy_assets( source_space, dest_space, dest_page, src.visual, src.body, note_asset, + attached=entries(src.meta), ) new_title = title if title is not None else (src.title or dest_page) if title is None and new_title == src.title and source_space == dest_space: @@ -774,7 +782,8 @@ def copy_page( dest_space, dest_page, title=new_title, abstractor=src.abstractor, visual=visual, body=body, expected_revision="", extra_meta={k: v for k, v in src.meta.items() - if k not in {"title", "abstractor", "visual", "space"}}, + if k not in {"title", "abstractor", "visual", "space", KEY}} + | ({KEY: json.dumps(attached)} if KEY in src.meta else {}), ) if record is not None: record["dest_revision"] = doc.revision diff --git a/coulomb_social/apps/spaces/urls.py b/coulomb_social/apps/spaces/urls.py index 41991f2..17b995f 100644 --- a/coulomb_social/apps/spaces/urls.py +++ b/coulomb_social/apps/spaces/urls.py @@ -12,6 +12,9 @@ urlpatterns = [ ), path("new/", views.space_create, name="create"), path("/assets/", views.space_asset, name="asset"), + path("/downloads/", views.space_download, name="download"), + path("/pages//attachments/", + views.page_attachments, name="page_attachments"), path("/edit/", views.space_edit, name="edit"), path("/sharing/", views.space_sharing, name="sharing"), path("/archive/", views.space_archive, name="archive"), diff --git a/coulomb_social/apps/spaces/views.py b/coulomb_social/apps/spaces/views.py index 2f2056a..7433e55 100644 --- a/coulomb_social/apps/spaces/views.py +++ b/coulomb_social/apps/spaces/views.py @@ -2,6 +2,7 @@ import hashlib import hmac import json import logging +from pathlib import PurePosixPath from django.conf import settings from django.contrib import messages @@ -21,8 +22,8 @@ from django.views.decorators.http import require_http_methods, require_POST, req from coulomb_social.apps.core.principal import build_principal -from . import pageops -from .assets import local_asset_url, open_local_image +from . import attachments, pageops +from .assets import local_asset_url, open_local_asset, open_local_image from .content import ( clear_content_cache_for_repo, clear_content_cache_for_space, @@ -79,6 +80,81 @@ def space_asset(request: HttpRequest, slug: str, asset_path: str) -> HttpRespons return response +@login_required +@require_safe +def space_download(request, slug, asset_path): + principal, forbid = _principal_or_forbid(request, "space_asset") + if forbid: + return forbid + space = get_space_for_member(principal.get("member"), slug) + if space is None or Space.objects.filter(slug=slug).exclude(pk=space.pk).exists(): + return HttpResponseNotFound("Attachment not found.") + try: + file = open_local_asset(slug, asset_path) + except (OSError, ValueError): + return HttpResponseNotFound("Attachment not found.") + response = FileResponse(file, as_attachment=True, filename=PurePosixPath(asset_path).name, + content_type="application/octet-stream") + response["Cache-Control"] = "private, no-store" + response["X-Content-Type-Options"] = "nosniff" + response["Content-Security-Policy"] = "default-src 'none'; sandbox" + return response + + +@login_required +@require_http_methods(["GET", "POST"]) +def page_attachments(request, slug, page_slug): + principal, forbid = _principal_or_forbid( + request, "page_edit" if request.method == "POST" else "space_detail") + if forbid: + return forbid + member = principal.get("member") + space = get_space_for_member(member, slug) + if space is None: + return HttpResponseNotFound("Page not found.") + can_write = can_write_space(member, space) + if request.method == "POST" and not can_write: + return HttpResponseForbidden("You have read-only access to this space.") + try: + doc = pageops.read_page(slug, page_slug) + if not doc.exists: + return HttpResponseNotFound("Page not found.") + if request.method == "POST": + try: + length = int(request.META.get("CONTENT_LENGTH") or 0) + except (ValueError, TypeError): + return HttpResponse("Invalid upload length.", status=400) + if length < 0: + return HttpResponse("Invalid upload length.", status=400) + if length > attachments.MAX_UPLOAD_BYTES + 65536: + return HttpResponse("Choose a file no larger than 16 MiB.", status=413) + action = request.POST.get("action") + if action == "add": + attachments.add(slug, page_slug, request.FILES.get("file"), + revision=request.POST.get("revision")) + messages.success(request, "Attachment added.") + elif action == "remove": + attachments.remove(slug, page_slug, request.POST.get("reference"), + revision=request.POST.get("revision")) + messages.success( + request, "Attachment removed from this page. Earlier revisions retain it.") + else: + raise pageops.PageOpsError("Choose an attachment action") + clear_content_cache_for_space(space) + return redirect("spaces:page_attachments", slug=slug, page_slug=page_slug) + files = attachments.page_files(slug, doc) + except (pageops.PageOpsError, OSError) as exc: + messages.error(request, _mutation_error(exc)) + if request.method == "POST": + return redirect("spaces:page_attachments", slug=slug, page_slug=page_slug) + return HttpResponse("Attachments unavailable.", status=409) + response = render(request, "spaces/attachments.html", { + "space": space, "doc": doc, "files": files, "can_write": can_write, + }) + response["Cache-Control"] = "private, no-store" + return response + + def _principal_or_forbid(request: HttpRequest, resource: str): principal = build_principal(request, authz_resource_id=resource) if not principal["authz_allow"]: diff --git a/coulomb_social/templates/spaces/attachments.html b/coulomb_social/templates/spaces/attachments.html new file mode 100644 index 0000000..5e74aa3 --- /dev/null +++ b/coulomb_social/templates/spaces/attachments.html @@ -0,0 +1,49 @@ +{% extends "base.html" %} +{% block title %}Attachments — {{ doc.title }}{% endblock %} +{% block content %} +

← {{ doc.title }}

+

Attachments

+

Files for {{ doc.title }}. Access follows this space’s sharing settings.

+
+
    + {% for file in files %} +
  • + {{ file.name }} + {% if file.open_url %}Open image{% endif %} + Download + {% if can_write %} + {% if file.managed %} +
    + {% csrf_token %} + + + + +
    + {% else %} + Edit page reference + {% endif %} + {% endif %} +
    • + {% empty %} +
  • No attachments on this page.
    • + {% endfor %} +
+
+{% if can_write %} +
+

Add attachment

+
+ {% csrf_token %} + + +


+

+ +
+

Removing a file from this page keeps it in earlier revisions. Existing links and visuals can be changed in the page editor.

+
+{% else %} +

Read-only access. Authors can add or remove attachments.

+{% endif %} +{% endblock %} diff --git a/coulomb_social/templates/spaces/detail.html b/coulomb_social/templates/spaces/detail.html index 14f3571..b61e246 100644 --- a/coulomb_social/templates/spaces/detail.html +++ b/coulomb_social/templates/spaces/detail.html @@ -109,6 +109,12 @@ title="Copy the page content to paste into an email" aria-controls="copy-content-status"> Copy content to clipboard + {% if page.source == 'content-plane' %} +
+ Attachments +

{% if can_write %}Open, download or manage attachments{% else %}Open or download attachments{% endif %}

+
+ {% endif %} {% endif %} {% if can_write %} Edit page diff --git a/scripts/test-attachments-browser.mjs b/scripts/test-attachments-browser.mjs new file mode 100644 index 0000000..ab7589d --- /dev/null +++ b/scripts/test-attachments-browser.mjs @@ -0,0 +1,94 @@ +import assert from 'node:assert/strict'; +import {chromium} from 'playwright'; + +const base = new URL(process.argv[2]); +assert.equal(base.protocol, 'http:'); +assert.ok(['localhost', '127.0.0.1'].includes(base.hostname)); +const browser = await chromium.launch({headless: true}); +try { + const context = await browser.newContext(); + const unexpected = []; + await context.route('**/*', route => { + if (new URL(route.request().url()).origin === base.origin) return route.continue(); + unexpected.push(route.request().url()); + return route.abort(); + }); + const page = await context.newPage(); + page.setDefaultTimeout(10000); + async function login(subject) { + await page.goto(`${base.origin}/auth/dev-login/`); + await page.getByLabel('Subject').fill(subject); + await page.getByLabel('Tenant').fill('tenant:attachment-test'); + await page.getByRole('button', {name: 'Establish session'}).click(); + await page.waitForURL('**/app/'); + } + const shared = `${base.origin}/app/spaces/attachments-shared/`; + const destination = `${base.origin}/app/spaces/attachments-private/`; + const source = `${shared}?page=notes`; + async function menu(url, author) { + await page.goto(url); + await page.locator('summary', {hasText: 'Attachments'}).click(); + await page.getByRole('link', {name: author + ? 'Open, download or manage attachments' : 'Open or download attachments', exact: true}).click(); + await page.waitForURL('**/attachments/'); + } + async function downloadedBytes() { + const pending = page.waitForEvent('download'); + await page.getByRole('link', {name: 'Download', exact: true}).click(); + const stream = await (await pending).createReadStream(); + const chunks = []; + for await (const chunk of stream) chunks.push(chunk); + return Buffer.concat(chunks).toString(); + } + await login('file-author'); + await menu(source, true); + await page.getByLabel('File (up to 16 MiB)').setInputFiles({ + name: 'meeting-notes.txt', mimeType: 'text/plain', buffer: Buffer.from('Synthetic attachment bytes'), + }); + await page.getByRole('button', {name: 'Add attachment', exact: true}).click(); + await page.waitForURL('**/attachments/'); + assert.match(await page.locator('main').innerText(), /Attachment added/); + const originalDownload = await page.getByRole('link', {name: 'Download', exact: true}).getAttribute('href'); + assert.equal(await downloadedBytes(), 'Synthetic attachment bytes'); + await login('file-reader'); + await menu(source, false); + assert.equal(await page.getByRole('button', {name: 'Add attachment', exact: true}).count(), 0); + assert.equal(await page.getByRole('button', {name: 'Remove from page', exact: true}).count(), 0); + assert.equal(await downloadedBytes(), 'Synthetic attachment bytes'); + await login('file-outsider'); + assert.equal((await page.goto(new URL(originalDownload, base).href)).status(), 404); + assert.equal((await page.goto(`${shared}pages/notes/attachments/`)).status(), 404); + await login('file-author'); + await page.goto(source); + await page.getByRole('link', {name: 'Duplicate page', exact: true}).click(); + await page.getByLabel('Destination space').selectOption('attachments-private'); + await page.getByLabel('Title', {exact: false}).fill('Independent files'); + await page.getByLabel('New page slug', {exact: false}).fill('copied'); + await page.getByRole('button', {name: 'Create duplicate'}).click(); + await page.waitForURL('**?page=copied'); + await menu(`${destination}?page=copied`, true); + assert.equal(await downloadedBytes(), 'Synthetic attachment bytes'); + const copiedDownload = await page.getByRole('link', {name: 'Download', exact: true}).getAttribute('href'); + assert.notEqual(copiedDownload, originalDownload); + await page.getByRole('button', {name: 'Remove from page', exact: true}).click(); + await page.waitForURL('**/attachments/'); + assert.match(await page.locator('main').innerText(), /No attachments on this page/); + await menu(source, true); + assert.equal(await downloadedBytes(), 'Synthetic attachment bytes'); + await page.goto(source); + await page.getByRole('link', {name: 'Transfer', exact: true}).click(); + await page.getByLabel('Destination space').selectOption('attachments-private'); + await page.getByLabel('Slug in destination', {exact: false}).fill('moved'); + await page.getByRole('button', {name: 'Transfer page', exact: true}).click(); + await page.waitForURL('**?page=moved'); + await menu(`${destination}?page=moved`, true); + assert.equal(await downloadedBytes(), 'Synthetic attachment bytes'); + const movedDownload = await page.getByRole('link', {name: 'Download', exact: true}).getAttribute('href'); + await login('file-reader'); + assert.equal((await page.goto(source)).status(), 404); + assert.equal((await page.goto(new URL(movedDownload, base).href)).status(), 404); + assert.deepEqual(unexpected, []); + console.log('Attachment menu upload, download, independent copy, move and removal passed'); +} finally { + await browser.close(); +} diff --git a/tests/test_attachments.py b/tests/test_attachments.py new file mode 100644 index 0000000..5533436 --- /dev/null +++ b/tests/test_attachments.py @@ -0,0 +1,239 @@ +"""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, '.html', b'') + 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'' + menu = client.get(reverse("spaces:page_attachments", args=[space.slug, doc.slug])) + assert b"