peter
e3c5773e8f
markdown/deck/html_fragment rendered author-influenced content verbatim,
so <script>/<img onerror>/etc. executed on the Daybook origin — self-XSS
against the owner's session, and (post-sharing) unauthenticated via /s/.
Two layers, both in artifacts/views.py:
- Create-time allowlist sanitization via nh3 (maintained ammonia/Rust
sanitizer, not deprecated bleach). Keeps formatting/tables/links/images
(incl. relative /assets URLs) and codehilite markup; drops script, on*
handlers, style, unknown tags, and javascript:/data: URLs. html_full is
deliberately NOT sanitized (documented verbatim escape hatch).
- Serve-time CSP (script-src 'none') for the sanitized kinds, on both
/a/{id} and the unauthenticated /s/{token} route. Being a response
header, it also neutralizes scripts in artifacts stored before this fix.
Not applied to html_full or json/csv (json/csv data is already escaped;
their viewers need their own JS).
Verified live across all kinds; 7 regression tests added. 102 tests pass.
Not addressed (noted in the issue): CSV formula injection,
CSRF_COOKIE_HTTPONLY (moot with XSS closed), nonce CSP for json/csv/full.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
596 lines
26 KiB
Python
596 lines
26 KiB
Python
import json
|
|
import shutil
|
|
import tempfile
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
|
|
import openpyxl
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import override_settings
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
from rest_framework.authtoken.models import Token
|
|
from rest_framework.test import APITestCase
|
|
|
|
from artifacts.models import Artifact
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
class ArtifactAPITestCase(APITestCase):
|
|
"""Exercises the ingest/list/detail API against a real (temp) filesystem
|
|
and a real (test) database, the way an agent posting over HTTP would."""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super().setUpClass()
|
|
cls.artifacts_root = Path(tempfile.mkdtemp(prefix="daybook-test-artifacts-"))
|
|
cls.settings_override = override_settings(ARTIFACTS_ROOT=cls.artifacts_root)
|
|
cls.settings_override.enable()
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.settings_override.disable()
|
|
shutil.rmtree(cls.artifacts_root, ignore_errors=True)
|
|
super().tearDownClass()
|
|
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(username="alice", password="pw")
|
|
self.token = Token.objects.create(user=self.user)
|
|
self.other_user = User.objects.create_user(username="bob", password="pw")
|
|
self.other_token = Token.objects.create(user=self.other_user)
|
|
|
|
def auth_headers(self, token):
|
|
return {"HTTP_AUTHORIZATION": f"Bearer {token.key}"}
|
|
|
|
def create_artifact(self, token=None, **overrides):
|
|
token = token or self.token
|
|
payload = {
|
|
"source": "test-agent",
|
|
"title": "Hello World",
|
|
"kind": "markdown",
|
|
"content": "# Hello\n\nBody text.",
|
|
}
|
|
payload.update(overrides)
|
|
return self.client.post(
|
|
reverse("artifact-api"), payload, format="json", **self.auth_headers(token)
|
|
)
|
|
|
|
# -- capabilities -------------------------------------------------
|
|
|
|
def test_capabilities_requires_auth(self):
|
|
response = self.client.get(reverse("capabilities-api"))
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
def test_capabilities_with_token(self):
|
|
response = self.client.get(reverse("capabilities-api"), **self.auth_headers(self.token))
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn("markdown", response.data["artifact_kinds"])
|
|
|
|
def test_capabilities_lists_supported_methods(self):
|
|
response = self.client.get(reverse("capabilities-api"), **self.auth_headers(self.token))
|
|
methods = response.data["methods"]
|
|
self.assertEqual(methods["/api/artifacts"], ["GET", "POST"])
|
|
self.assertEqual(methods["/api/artifacts/{id}"], ["GET", "PATCH", "DELETE"])
|
|
|
|
def test_capabilities_with_session(self):
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("capabilities-api"))
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
# -- create (POST) --------------------------------------------------
|
|
|
|
def test_create_markdown_artifact(self):
|
|
response = self.create_artifact(tags=["demo", "weekly"])
|
|
self.assertEqual(response.status_code, 201)
|
|
body = response.data
|
|
self.assertEqual(body["source"], "test-agent")
|
|
self.assertEqual(body["kind"], "markdown")
|
|
self.assertEqual(body["title"], "Hello World")
|
|
self.assertEqual([tag["name"] for tag in body["tags"]], ["demo", "weekly"])
|
|
|
|
artifact = Artifact.objects.get(pk=body["id"])
|
|
self.assertEqual(artifact.owner, self.user)
|
|
|
|
def test_create_artifact_writes_files_to_disk(self):
|
|
response = self.create_artifact()
|
|
artifact = Artifact.objects.get(pk=response.data["id"])
|
|
stored_path = self.artifacts_root / artifact.path
|
|
self.assertTrue(stored_path.exists())
|
|
# Markdown artifacts also preserve the original .md alongside the rendered .html
|
|
self.assertTrue(stored_path.with_suffix(".md").exists())
|
|
|
|
def test_create_artifact_requires_token(self):
|
|
response = self.client.post(
|
|
reverse("artifact-api"),
|
|
{"source": "test-agent", "title": "x", "kind": "markdown", "content": "hi"},
|
|
format="json",
|
|
)
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
def test_create_artifact_rejects_invalid_kind(self):
|
|
response = self.create_artifact(kind="pdf")
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_create_artifact_rejects_missing_fields(self):
|
|
response = self.client.post(
|
|
reverse("artifact-api"), {"kind": "markdown"}, format="json", **self.auth_headers(self.token)
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
# -- list (GET) -------------------------------------------------------
|
|
|
|
def test_list_requires_session_login(self):
|
|
response = self.client.get(reverse("artifact-api"))
|
|
self.assertEqual(response.status_code, 302) # redirected to login
|
|
|
|
def test_list_scoped_to_owner(self):
|
|
self.create_artifact()
|
|
self.create_artifact(token=self.other_token, title="Not mine")
|
|
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("artifact-api"))
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual([a["title"] for a in response.data], ["Hello World"])
|
|
|
|
def test_list_filter_by_label(self):
|
|
self.create_artifact(tags=["urgent"])
|
|
self.create_artifact(title="Second", tags=["later"])
|
|
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("artifact-api"), {"label": "urgent"})
|
|
self.assertEqual(len(response.data), 1)
|
|
self.assertEqual(response.data[0]["title"], "Hello World")
|
|
|
|
def test_list_filter_by_kind(self):
|
|
self.create_artifact(kind="markdown")
|
|
self.create_artifact(title="Frag", kind="html_fragment", content="<p>hi</p>")
|
|
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("artifact-api"), {"kind": "html_fragment"})
|
|
self.assertEqual(len(response.data), 1)
|
|
self.assertEqual(response.data[0]["title"], "Frag")
|
|
|
|
def test_list_filter_by_query(self):
|
|
self.create_artifact(title="Weekly Report")
|
|
self.create_artifact(title="Random Notes")
|
|
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("artifact-api"), {"q": "weekly"})
|
|
self.assertEqual(len(response.data), 1)
|
|
self.assertEqual(response.data[0]["title"], "Weekly Report")
|
|
|
|
def test_list_filter_by_day(self):
|
|
self.create_artifact()
|
|
today = timezone.localdate().isoformat()
|
|
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("artifact-api"), {"day": today})
|
|
self.assertEqual(len(response.data), 1)
|
|
|
|
response = self.client.get(reverse("artifact-api"), {"day": "2000-01-01"})
|
|
self.assertEqual(len(response.data), 0)
|
|
|
|
# -- detail (GET/PATCH) -----------------------------------------------
|
|
|
|
def test_detail_get_scoped_to_owner(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
|
|
self.client.force_login(self.other_user)
|
|
response = self.client.get(reverse("artifact-detail-api", args=[artifact_id]))
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("artifact-detail-api", args=[artifact_id]))
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_patch_updates_title_and_tags(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
response = self.client.patch(
|
|
reverse("artifact-detail-api", args=[artifact_id]),
|
|
{"title": "Updated title", "tags": ["a", "b"]},
|
|
format="json",
|
|
**self.auth_headers(self.token),
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data["title"], "Updated title")
|
|
self.assertEqual([t["name"] for t in response.data["tags"]], ["a", "b"])
|
|
|
|
def test_patch_rejects_blank_title(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
response = self.client.patch(
|
|
reverse("artifact-detail-api", args=[artifact_id]),
|
|
{"title": ""},
|
|
format="json",
|
|
**self.auth_headers(self.token),
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_patch_requires_token(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
response = self.client.patch(
|
|
reverse("artifact-detail-api", args=[artifact_id]), {"title": "No token"}, format="json"
|
|
)
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
def test_patch_cannot_touch_other_users_artifact(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
response = self.client.patch(
|
|
reverse("artifact-detail-api", args=[artifact_id]),
|
|
{"title": "Hijacked"},
|
|
format="json",
|
|
**self.auth_headers(self.other_token),
|
|
)
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
# -- delete (DELETE) --------------------------------------------------
|
|
|
|
def test_delete_removes_artifact_and_files(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
artifact = Artifact.objects.get(pk=artifact_id)
|
|
stored_path = self.artifacts_root / artifact.path
|
|
self.assertTrue(stored_path.exists())
|
|
|
|
response = self.client.delete(
|
|
reverse("artifact-detail-api", args=[artifact_id]), **self.auth_headers(self.token)
|
|
)
|
|
self.assertEqual(response.status_code, 204)
|
|
self.assertFalse(Artifact.objects.filter(pk=artifact_id).exists())
|
|
self.assertFalse(stored_path.exists())
|
|
self.assertFalse(stored_path.with_suffix(".md").exists())
|
|
|
|
def test_delete_requires_token(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
response = self.client.delete(reverse("artifact-detail-api", args=[artifact_id]))
|
|
self.assertEqual(response.status_code, 403)
|
|
self.assertTrue(Artifact.objects.filter(pk=artifact_id).exists())
|
|
|
|
def test_delete_cannot_touch_other_users_artifact(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
response = self.client.delete(
|
|
reverse("artifact-detail-api", args=[artifact_id]), **self.auth_headers(self.other_token)
|
|
)
|
|
self.assertEqual(response.status_code, 404)
|
|
self.assertTrue(Artifact.objects.filter(pk=artifact_id).exists())
|
|
|
|
def test_delete_twice_second_call_404s(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
first = self.client.delete(
|
|
reverse("artifact-detail-api", args=[artifact_id]), **self.auth_headers(self.token)
|
|
)
|
|
self.assertEqual(first.status_code, 204)
|
|
|
|
second = self.client.delete(
|
|
reverse("artifact-detail-api", args=[artifact_id]), **self.auth_headers(self.token)
|
|
)
|
|
self.assertEqual(second.status_code, 404)
|
|
|
|
def test_deleted_artifact_is_gone_from_list_and_detail(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
self.client.delete(
|
|
reverse("artifact-detail-api", args=[artifact_id]), **self.auth_headers(self.token)
|
|
)
|
|
|
|
self.client.force_login(self.user)
|
|
list_response = self.client.get(reverse("artifact-api"))
|
|
self.assertEqual(list_response.data, [])
|
|
|
|
detail_response = self.client.get(reverse("artifact-detail-api", args=[artifact_id]))
|
|
self.assertEqual(detail_response.status_code, 404)
|
|
|
|
# -- field length limits (issue #003) ----------------------------------
|
|
|
|
def test_create_rejects_oversized_title(self):
|
|
response = self.create_artifact(title="A" * 201)
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertIn("title", response.data["detail"])
|
|
|
|
def test_create_rejects_oversized_source(self):
|
|
response = self.create_artifact(source="a" * 81)
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertIn("source", response.data["detail"])
|
|
|
|
def test_create_accepts_title_at_exact_limit(self):
|
|
response = self.create_artifact(title="A" * 200)
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
def test_patch_rejects_oversized_title(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
response = self.client.patch(
|
|
reverse("artifact-detail-api", args=[artifact_id]),
|
|
{"title": "A" * 201},
|
|
format="json",
|
|
**self.auth_headers(self.token),
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
# -- content-shape validation (issue #004) -----------------------------
|
|
|
|
def test_create_rejects_nested_html_in_html_fragment(self):
|
|
response = self.create_artifact(
|
|
kind="html_fragment", content="<html><body>nested</body></html>"
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_create_rejects_html_full_without_html_tag(self):
|
|
response = self.create_artifact(kind="html_full", content="<p>not a full document</p>")
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_create_rejects_deck_without_section(self):
|
|
response = self.create_artifact(kind="deck", content="<h1>Not a deck slide</h1>")
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_create_rejects_whitespace_only_markdown(self):
|
|
response = self.create_artifact(kind="markdown", content=" \n\t ")
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_create_accepts_well_formed_html_fragment(self):
|
|
response = self.create_artifact(kind="html_fragment", content="<p>fine</p>")
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
# -- PATCH validation (issue #005) -------------------------------------
|
|
|
|
def test_patch_rejects_non_string_non_list_tags(self):
|
|
artifact_id = self.create_artifact(tags=["important", "keep-me"]).data["id"]
|
|
response = self.client.patch(
|
|
reverse("artifact-detail-api", args=[artifact_id]),
|
|
{"tags": 12345},
|
|
format="json",
|
|
**self.auth_headers(self.token),
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
artifact = Artifact.objects.get(pk=artifact_id)
|
|
self.assertEqual(artifact.tags, ["important", "keep-me"])
|
|
|
|
def test_patch_rejects_immutable_fields(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
for field, value in [
|
|
("content", "new content"),
|
|
("source", "other-source"),
|
|
("kind", "deck"),
|
|
("day", "2020-01-01"),
|
|
]:
|
|
response = self.client.patch(
|
|
reverse("artifact-detail-api", args=[artifact_id]),
|
|
{field: value},
|
|
format="json",
|
|
**self.auth_headers(self.token),
|
|
)
|
|
self.assertEqual(response.status_code, 400, field)
|
|
self.assertIn(field, response.data["detail"])
|
|
|
|
# -- syntax highlighting -------------------------------------------------
|
|
|
|
def test_markdown_fenced_code_block_gets_syntax_highlighted(self):
|
|
response = self.create_artifact(
|
|
content="# Test\n\n```python\ndef hello():\n return \"world\"\n```"
|
|
)
|
|
artifact = Artifact.objects.get(pk=response.data["id"])
|
|
stored_html = (self.artifacts_root / artifact.path).read_text(encoding="utf-8")
|
|
|
|
self.assertIn('class="codehilite"', stored_html)
|
|
self.assertIn('<span class="k">def</span>', stored_html)
|
|
self.assertIn('href="http://testserver/static/themes/pygments.css"', stored_html)
|
|
|
|
# -- json kind ------------------------------------------------------------
|
|
|
|
def test_create_json_artifact(self):
|
|
response = self.create_artifact(kind="json", content='{"a": 1, "b": [1, 2, 3]}')
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
artifact = Artifact.objects.get(pk=response.data["id"])
|
|
stored_html = (self.artifacts_root / artifact.path).read_text(encoding="utf-8")
|
|
self.assertIn('renderjson/renderjson.js', stored_html)
|
|
self.assertIn('type="application/json"', stored_html)
|
|
|
|
# raw sidecar preserves the exact original JSON text
|
|
raw_path = (self.artifacts_root / artifact.path).with_suffix(".json")
|
|
self.assertEqual(raw_path.read_text(encoding="utf-8"), '{"a": 1, "b": [1, 2, 3]}')
|
|
|
|
def test_create_json_artifact_rejects_invalid_json(self):
|
|
response = self.create_artifact(kind="json", content="{not valid json")
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_json_artifact_escapes_script_close_tag_safely(self):
|
|
payload = json.dumps({"value": "</script><script>alert(1)</script>"})
|
|
response = self.create_artifact(kind="json", content=payload)
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
artifact = Artifact.objects.get(pk=response.data["id"])
|
|
stored_html = (self.artifacts_root / artifact.path).read_text(encoding="utf-8")
|
|
|
|
# The page has exactly 3 real <script> elements (renderjson lib, the
|
|
# embedded data block, the init snippet). If the injected payload's
|
|
# "</script" wasn't escaped, it would add a 4th closing tag here.
|
|
self.assertEqual(stored_html.count("</script>"), 3)
|
|
|
|
# The escaped content must still round-trip to the exact original value.
|
|
start = stored_html.index('id="daybook-data">') + len('id="daybook-data">')
|
|
end = stored_html.index("</script>", start)
|
|
embedded = stored_html[start:end]
|
|
self.assertEqual(json.loads(embedded)["value"], "</script><script>alert(1)</script>")
|
|
|
|
# -- csv kind ---------------------------------------------------------------
|
|
|
|
def test_create_csv_artifact(self):
|
|
response = self.create_artifact(kind="csv", content="name,score\nalice,10\nbob,7")
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
artifact = Artifact.objects.get(pk=response.data["id"])
|
|
stored_html = (self.artifacts_root / artifact.path).read_text(encoding="utf-8")
|
|
self.assertIn("gridjs/gridjs.umd.js", stored_html)
|
|
self.assertIn('"columns": ["name", "score"]', stored_html)
|
|
self.assertIn('["alice", "10"]', stored_html)
|
|
|
|
raw_path = (self.artifacts_root / artifact.path).with_suffix(".csv")
|
|
self.assertEqual(raw_path.read_text(encoding="utf-8"), "name,score\nalice,10\nbob,7")
|
|
|
|
def test_create_csv_artifact_with_header_only(self):
|
|
response = self.create_artifact(kind="csv", content="name,score")
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
# -- source_name/source_id/source_prompt type checks (issue #008) -------
|
|
|
|
def test_create_rejects_non_string_source_name(self):
|
|
response = self.create_artifact(source_name=12345)
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertIn("source_name", response.data["detail"])
|
|
|
|
def test_create_rejects_non_string_source_id(self):
|
|
response = self.create_artifact(source_id=["a", "b"])
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_create_rejects_non_string_source_prompt(self):
|
|
response = self.create_artifact(source_prompt={"nested": "object"})
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_patch_rejects_non_string_source_name(self):
|
|
artifact_id = self.create_artifact().data["id"]
|
|
response = self.client.patch(
|
|
reverse("artifact-detail-api", args=[artifact_id]),
|
|
{"source_name": 12345},
|
|
format="json",
|
|
**self.auth_headers(self.token),
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
artifact = Artifact.objects.get(pk=artifact_id)
|
|
self.assertEqual(artifact.source_name, "test-agent") # unchanged, not mangled
|
|
|
|
# -- xlsx upload converts to csv kind ------------------------------------
|
|
|
|
def make_xlsx(self, rows, filename="sheet.xlsx"):
|
|
workbook = openpyxl.Workbook()
|
|
sheet = workbook.active
|
|
for row in rows:
|
|
sheet.append(row)
|
|
buffer = BytesIO()
|
|
workbook.save(buffer)
|
|
buffer.seek(0)
|
|
buffer.name = filename
|
|
return buffer
|
|
|
|
def test_xlsx_upload_converts_to_csv_kind(self):
|
|
xlsx = self.make_xlsx([["Item", "Q1", "Q2"], ["Cash", 1000, 1200]])
|
|
response = self.client.post(
|
|
reverse("artifact-api"),
|
|
{"source": "test-agent", "title": "Balance Sheet", "file": xlsx},
|
|
format="multipart",
|
|
**self.auth_headers(self.token),
|
|
)
|
|
self.assertEqual(response.status_code, 201)
|
|
self.assertEqual(response.data["kind"], "csv")
|
|
|
|
artifact = Artifact.objects.get(pk=response.data["id"])
|
|
raw_path = (self.artifacts_root / artifact.path).with_suffix(".csv")
|
|
# read_text() normalizes line endings, so this is OS-portable rather
|
|
# than asserting the exact on-disk bytes (which are "\r\n" on Windows).
|
|
self.assertEqual(raw_path.read_text(encoding="utf-8"), "Item,Q1,Q2\nCash,1000,1200\n")
|
|
|
|
stored_html = (self.artifacts_root / artifact.path).read_text(encoding="utf-8")
|
|
self.assertIn('"columns": ["Item", "Q1", "Q2"]', stored_html)
|
|
self.assertIn('["Cash", "1000", "1200"]', stored_html)
|
|
|
|
def test_xlsx_upload_does_not_mangle_fields(self):
|
|
"""Regression test: {**a_querydict} previously wrapped every value
|
|
(including title) in a list - QueryDict's __getitem__ override is
|
|
bypassed by the ** spread, exposing its internal list-per-key
|
|
storage directly. Separately, pre-setting None for absent optional
|
|
fields defeated create_text_artifact's own .get(field, default)
|
|
calls (.get only falls through when the key is missing, not None),
|
|
which previously caused a NOT NULL constraint failure on
|
|
source_prompt."""
|
|
xlsx = self.make_xlsx([["a", "b"]])
|
|
response = self.client.post(
|
|
reverse("artifact-api"),
|
|
{
|
|
"source": "test-agent",
|
|
"title": "Untitled Sheet",
|
|
"tags": "finance,q2",
|
|
"file": xlsx,
|
|
},
|
|
format="multipart",
|
|
**self.auth_headers(self.token),
|
|
)
|
|
self.assertEqual(response.status_code, 201)
|
|
self.assertEqual(response.data["title"], "Untitled Sheet") # not ["Untitled Sheet"]
|
|
self.assertEqual(response.data["source_prompt"], "") # defaulted, not None
|
|
self.assertEqual([t["name"] for t in response.data["tags"]], ["finance", "q2"])
|
|
|
|
def test_xlsx_upload_rejects_corrupt_file(self):
|
|
buffer = BytesIO(b"not a real xlsx file")
|
|
buffer.name = "broken.xlsx"
|
|
response = self.client.post(
|
|
reverse("artifact-api"),
|
|
{"source": "test-agent", "title": "Broken", "file": buffer},
|
|
format="multipart",
|
|
**self.auth_headers(self.token),
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
# -- XSS sanitization of themed kinds (issue #014) -----------------------
|
|
|
|
def _stored_html(self, artifact_id):
|
|
artifact = Artifact.objects.get(pk=artifact_id)
|
|
return (self.artifacts_root / artifact.path).read_text(encoding="utf-8")
|
|
|
|
def test_markdown_strips_script_and_event_handlers(self):
|
|
aid = self.create_artifact(
|
|
kind="markdown",
|
|
content="# Hi\n\n<script>alert(document.cookie)</script>\n\n<img src=x onerror=alert(1)>\n\n[x](javascript:alert(1))",
|
|
).data["id"]
|
|
html = self._stored_html(aid).lower()
|
|
self.assertNotIn("<script", html)
|
|
self.assertNotIn("onerror", html)
|
|
self.assertNotIn("javascript:", html)
|
|
|
|
def test_html_fragment_strips_script_and_svg_onload(self):
|
|
aid = self.create_artifact(
|
|
kind="html_fragment",
|
|
content="<p>hi</p><script>alert(1)</script><svg onload=alert(2)></svg>",
|
|
).data["id"]
|
|
html = self._stored_html(aid).lower()
|
|
self.assertNotIn("<script", html)
|
|
self.assertNotIn("onload", html)
|
|
self.assertNotIn("<svg", html)
|
|
|
|
def test_deck_strips_script(self):
|
|
aid = self.create_artifact(
|
|
kind="deck", content="<section>Slide<script>alert(1)</script></section>"
|
|
).data["id"]
|
|
self.assertNotIn("<script", self._stored_html(aid).lower())
|
|
|
|
def test_sanitization_preserves_legitimate_formatting(self):
|
|
aid = self.create_artifact(
|
|
kind="markdown",
|
|
content="# Title\n\n| a | b |\n|---|---|\n| 1 | 2 |\n\n```python\ndef f():\n return 1\n```\n\n",
|
|
).data["id"]
|
|
html = self._stored_html(aid)
|
|
self.assertIn("<table>", html)
|
|
self.assertIn('class="codehilite"', html)
|
|
self.assertIn('<span class="k">def</span>', html)
|
|
self.assertIn("/assets/ab12cd34ef56", html) # relative image URL survives
|
|
|
|
def test_html_full_is_not_sanitized(self):
|
|
aid = self.create_artifact(
|
|
kind="html_full",
|
|
content="<!doctype html><html><head></head><body><script>ok()</script></body></html>",
|
|
).data["id"]
|
|
# html_full is the deliberate verbatim escape hatch — its script stays.
|
|
self.assertIn("<script>ok()</script>", self._stored_html(aid))
|
|
|
|
def test_sanitized_kinds_served_with_script_blocking_csp(self):
|
|
aid = self.create_artifact(kind="markdown", content="# hi").data["id"]
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("artifact", args=[aid]))
|
|
self.assertIn("script-src 'none'", response.headers.get("Content-Security-Policy", ""))
|
|
|
|
def test_html_full_served_without_restrictive_csp(self):
|
|
aid = self.create_artifact(
|
|
kind="html_full", content="<!doctype html><html><body>x</body></html>"
|
|
).data["id"]
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("artifact", args=[aid]))
|
|
self.assertIsNone(response.headers.get("Content-Security-Policy"))
|
|
|