Serve artifacts of a type that can run script with CSP sandbox and nosniff
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013GAeg4Jo6xjZzjwy9QDWMQ
This commit is contained in:
@@ -27,6 +27,11 @@ from fluksio.flow.artifacts import ArtifactStore
|
||||
#: trusting one — anything else is served as bytes.
|
||||
_MEDIA_TYPE = re.compile(r"^[\w.+-]+/[\w.+-]+$")
|
||||
|
||||
#: What a browser shows without running anything, so it is served as it is.
|
||||
#: Every other type is served sandboxed: HTML, SVG and XML run script when
|
||||
#: opened as a page, and so may a type nobody listed here.
|
||||
_INERT = re.compile(r"^(audio|video)/|^image/(?!svg)|^application/(pdf|json)$")
|
||||
|
||||
|
||||
def artifact_caller(request: Request) -> str:
|
||||
"""Who may move artifacts: a signed-in person, or an attached worker.
|
||||
@@ -137,11 +142,14 @@ def get_artifact(
|
||||
path = store.path(digest)
|
||||
if path is None:
|
||||
raise HTTPException(status_code=404, detail="No such artifact")
|
||||
return FileResponse(
|
||||
path,
|
||||
media_type=(
|
||||
media_type if _MEDIA_TYPE.match(media_type) else "application/octet-stream"
|
||||
),
|
||||
# The digest is the content, so it is also the perfect validator.
|
||||
headers={"ETag": f'"{digest}"'},
|
||||
)
|
||||
if not _MEDIA_TYPE.match(media_type):
|
||||
media_type = "application/octet-stream"
|
||||
# The digest is the content, so it is also the perfect validator.
|
||||
headers = {"ETag": f'"{digest}"'}
|
||||
if not _INERT.match(media_type.lower()):
|
||||
# The caller names the type and any member can upload, so an artifact
|
||||
# asked for as text/html would otherwise run its script on this origin,
|
||||
# beside the session, for whoever follows a link to it.
|
||||
headers["Content-Security-Policy"] = "sandbox"
|
||||
headers["X-Content-Type-Options"] = "nosniff"
|
||||
return FileResponse(path, media_type=media_type, headers=headers)
|
||||
|
||||
@@ -180,3 +180,29 @@ def test_an_upload_may_ask_for_the_ring(
|
||||
assert served.content == b"\x89PNG-frame"
|
||||
finally:
|
||||
store.volatile = previous
|
||||
|
||||
|
||||
def test_a_type_that_can_run_script_is_served_sandboxed(
|
||||
client: TestClient, superuser_token_headers: dict[str, str]
|
||||
) -> None:
|
||||
"""Any member can upload, and the caller names the type it is served as."""
|
||||
url = f"{settings.API_V1_STR}/artifacts"
|
||||
page = client.put(
|
||||
f"{url}?name=x.html", headers=superuser_token_headers, content=b"<script>"
|
||||
).json()["digest"]
|
||||
image = client.put(
|
||||
f"{url}?name=x.png", headers=superuser_token_headers, content=b"\x89PNG"
|
||||
).json()["digest"]
|
||||
|
||||
def served(digest: str, media_type: str) -> dict[str, str]:
|
||||
answer = client.get(
|
||||
f"{url}/{digest}?media_type={media_type}", headers=superuser_token_headers
|
||||
)
|
||||
assert answer.status_code == 200
|
||||
return dict(answer.headers)
|
||||
|
||||
for media_type in ("text/html", "image/svg+xml", "application/xml"):
|
||||
headers = served(page, media_type)
|
||||
assert headers["content-security-policy"] == "sandbox"
|
||||
assert headers["x-content-type-options"] == "nosniff"
|
||||
assert "content-security-policy" not in served(image, "image/png")
|
||||
|
||||
Reference in New Issue
Block a user