mirror of
https://github.com/ArchiveBox/ArchiveBox.git
synced 2026-09-12 19:50:57 +05:00
fix: secure snapshot output deletion handoff
This commit is contained in:
Nick Sweeting
parent
b5560f77ff
commit
c8ef576235
@ -10,9 +10,12 @@ from pathlib import Path
|
||||
from urllib.parse import quote
|
||||
|
||||
from django.contrib import admin
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.contrib.admin.actions import delete_selected
|
||||
from django.contrib.admin.helpers import ACTION_CHECKBOX_NAME
|
||||
from django.core.exceptions import PermissionDenied, SuspiciousOperation, ValidationError
|
||||
from django.db.models import Count, Min, Prefetch, Q, Subquery, TextField, Window
|
||||
from django.db.models.functions import Cast
|
||||
from django.shortcuts import redirect
|
||||
from django.urls import resolve, reverse
|
||||
from django.utils import timezone
|
||||
from django.utils.html import format_html
|
||||
@ -543,10 +546,44 @@ class ArchiveResultAdmin(BaseModelAdmin):
|
||||
|
||||
def changelist_view(self, request, extra_context=None):
|
||||
self.request = request
|
||||
selected = request.GET.getlist(ACTION_CHECKBOX_NAME)
|
||||
if request.method == "GET" and request.GET.get("action") == "delete_selected" and selected:
|
||||
if not request.user.is_superuser:
|
||||
raise PermissionDenied
|
||||
if len(selected) > 100:
|
||||
raise SuspiciousOperation("Too many ArchiveResults selected for deletion")
|
||||
try:
|
||||
queryset = self.get_queryset(request).filter(pk__in=selected)
|
||||
if not queryset.exists():
|
||||
snapshot = Snapshot.objects.only("id").filter(pk=request.GET.get("snapshot")).first()
|
||||
return redirect(build_snapshot_url(str(snapshot.id), "index.html", request=request) if snapshot else request.path)
|
||||
except (ValidationError, ValueError):
|
||||
return redirect(request.path)
|
||||
return delete_selected(self, request, queryset)
|
||||
handoff_snapshot = (
|
||||
request.GET.get("snapshot") if request.method == "POST" and request.GET.get("action") == "delete_selected" else None
|
||||
)
|
||||
if handoff_snapshot:
|
||||
request.GET = request.GET.copy()
|
||||
request.GET.clear()
|
||||
request.META["QUERY_STRING"] = ""
|
||||
try:
|
||||
handoff_snapshot = Snapshot.objects.only("id").filter(pk=handoff_snapshot).first()
|
||||
except (ValidationError, ValueError):
|
||||
handoff_snapshot = None
|
||||
saved_list_per_page = self.list_per_page
|
||||
self.list_per_page = request.archivebox_config.SNAPSHOTS_PER_PAGE
|
||||
try:
|
||||
return super().changelist_view(request, extra_context)
|
||||
response = super().changelist_view(request, extra_context)
|
||||
if (
|
||||
handoff_snapshot
|
||||
and response.status_code in (301, 302)
|
||||
and not ArchiveResult.objects.filter(
|
||||
pk__in=request.POST.getlist(ACTION_CHECKBOX_NAME),
|
||||
).exists()
|
||||
):
|
||||
return redirect(build_snapshot_url(str(handoff_snapshot.id), "index.html", request=request))
|
||||
return response
|
||||
finally:
|
||||
self.list_per_page = saved_list_per_page
|
||||
|
||||
|
||||
@ -39,8 +39,8 @@ ADMIN_LOGIN_HINT_COOKIE = "archivebox_admin_logged_in"
|
||||
def _admin_login_hint_cookie_domain(config) -> str | None:
|
||||
"""Resolve the parent domain to scope the cross-subdomain login hint.
|
||||
|
||||
NOTE: this cookie carries only the single bit "user is logged in on
|
||||
admin somewhere"; it MUST NOT be confused with the session cookie,
|
||||
NOTE: this cookie carries only the single bit "a superuser is logged in
|
||||
on admin somewhere"; it MUST NOT be confused with the session cookie,
|
||||
which stays admin-host-scoped (see core/settings.py
|
||||
SESSION_COOKIE_DOMAIN comment — admin/web is a security boundary).
|
||||
|
||||
@ -88,6 +88,10 @@ def AdminCookieIsolationMiddleware(get_response):
|
||||
def middleware(request):
|
||||
response = get_response(request)
|
||||
|
||||
if request.path == "/admin" or request.path.startswith("/admin/"):
|
||||
response.headers["X-Frame-Options"] = "DENY"
|
||||
response.headers["Content-Security-Policy"] = "frame-ancestors 'none'"
|
||||
|
||||
config = request.__dict__.get("archivebox_config")
|
||||
if config is None or config.SERVER_SECURITY_MODE == "auto":
|
||||
from archivebox.config.common import get_request_config
|
||||
@ -197,6 +201,21 @@ def ServerSecurityModeMiddleware(get_response):
|
||||
|
||||
config = get_request_config(request, resolve_plugins=False)
|
||||
|
||||
if config.USES_SUBDOMAIN_ROUTING and config.BASE_URL and request.method.upper() not in allowed_methods:
|
||||
request_host, _request_port = split_host_port((request.get_host() or "").lower())
|
||||
control_hosts = {
|
||||
split_host_port(host)[0]
|
||||
for host in (
|
||||
get_base_host(config=config),
|
||||
get_admin_host(config=config),
|
||||
get_api_host(config=config),
|
||||
get_web_host(config=config),
|
||||
)
|
||||
if host
|
||||
}
|
||||
if request_host not in control_hosts:
|
||||
return HttpResponseForbidden("ArchiveBox is running with the control plane disabled on this host.")
|
||||
|
||||
if config.CONTROL_PLANE_ENABLED:
|
||||
return get_response(request)
|
||||
|
||||
@ -319,7 +338,12 @@ def HostRoutingMiddleware(get_response):
|
||||
return redirect(target)
|
||||
response = get_response(request)
|
||||
hint_cookie_domain = _admin_login_hint_cookie_domain(config)
|
||||
if request.user.is_authenticated and not request.path.startswith("/admin/logout"):
|
||||
if (
|
||||
request.user.is_authenticated
|
||||
and request.user.is_active
|
||||
and request.user.is_superuser
|
||||
and not request.path.startswith("/admin/logout")
|
||||
):
|
||||
response.set_cookie(
|
||||
ADMIN_LOGIN_HINT_COOKIE,
|
||||
"1",
|
||||
|
||||
@ -3594,6 +3594,14 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
|
||||
"warc/",
|
||||
)
|
||||
user = getattr(request, "user", None)
|
||||
can_delete_outputs = bool(
|
||||
static_export_dir is None
|
||||
and request is not None
|
||||
and (
|
||||
(user and user.is_authenticated and user.is_active and user.is_superuser)
|
||||
or request.COOKIES.get("archivebox_admin_logged_in") == "1"
|
||||
),
|
||||
)
|
||||
tag_widget = TagEditorWidget()
|
||||
return {
|
||||
"id": str(self.id),
|
||||
@ -3626,7 +3634,7 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
|
||||
"related_years": related_years,
|
||||
"loose_items": loose_items,
|
||||
"failed_items": failed_items,
|
||||
"can_delete_outputs": bool(user and user.is_authenticated and user.is_active and user.is_superuser),
|
||||
"can_delete_outputs": can_delete_outputs,
|
||||
"title_tags": [{"name": tag.name, "style": tag_widget._tag_style(tag.name)} for tag in sorted(tags, key=lambda tag: tag.name)],
|
||||
"STATIC_EXPORT": static_export_dir is not None,
|
||||
"STATIC_EXPORT_DIR": static_export_dir,
|
||||
|
||||
@ -1732,7 +1732,7 @@
|
||||
<a href="{{display_url}}" data-no-preview="1" title="Download output file" download>⬇️</a>
|
||||
{% endif %}
|
||||
{% if can_delete_outputs and result.result %}
|
||||
<button type="button" data-no-preview="1" data-archive-result-ids="{{result.result_ids}}" data-delete-url="{% admin_base_url %}/admin/core/archiveresult/" title="Delete this output">×</button>
|
||||
<button type="button" data-no-preview="1" data-archive-result-ids="{{result.result_ids}}" data-delete-url="{% admin_base_url %}/admin/core/archiveresult/" data-delete-handoff="1" data-delete-snapshot-id="{{snapshot.id}}" title="Delete this output">×</button>
|
||||
{% endif %}
|
||||
</div>
|
||||
{% if display_path %}
|
||||
@ -1819,7 +1819,6 @@
|
||||
</div>
|
||||
</header>
|
||||
|
||||
{% if can_delete_outputs %}<input type="hidden" id="delete-output-csrf" value="{{csrf_token}}">{% endif %}
|
||||
{% if can_delete_outputs %}{% include "includes/output_delete_controls.html" %}{% endif %}
|
||||
|
||||
<script>
|
||||
|
||||
@ -55,10 +55,20 @@
|
||||
}
|
||||
|
||||
function deleteQueuedOutputs() {
|
||||
const csrf = document.querySelector('#delete-output-csrf, input[name="csrfmiddlewaretoken"]')?.value
|
||||
const deleteUrl = deleteOutputButtons.find((button) => button.dataset.deleteUrl)?.dataset.deleteUrl
|
||||
if (!csrf || !deleteUrl) return window.alert('Delete failed: missing admin request data')
|
||||
const button = deleteOutputButtons.find((candidate) => candidate.dataset.deleteUrl)
|
||||
let deleteUrl = button?.dataset.deleteUrl
|
||||
if (!deleteUrl) return window.alert('Delete failed: missing admin request data')
|
||||
deleteOutputButtons.forEach((button) => button.disabled = true)
|
||||
if (button.dataset.deleteHandoff) {
|
||||
deleteUrl = new URL(deleteUrl, window.location.href)
|
||||
deleteUrl.searchParams.set('action', 'delete_selected')
|
||||
deleteUrl.searchParams.set('snapshot', button.dataset.deleteSnapshotId)
|
||||
queuedOutputIds.forEach((id) => deleteUrl.searchParams.append('_selected_action', id))
|
||||
window.location.assign(deleteUrl)
|
||||
return
|
||||
}
|
||||
const csrf = document.querySelector('input[name="csrfmiddlewaretoken"]')?.value
|
||||
if (!csrf) return window.alert('Delete failed: missing admin request data')
|
||||
const body = new URLSearchParams({action: 'delete_selected', post: 'yes', csrfmiddlewaretoken: csrf})
|
||||
queuedOutputIds.forEach((id) => body.append('_selected_action', id))
|
||||
fetch(deleteUrl, {
|
||||
|
||||
@ -16,6 +16,7 @@ from django.core.paginator import UnorderedObjectListWarning
|
||||
from django.test import RequestFactory
|
||||
from django.urls import reverse
|
||||
|
||||
from archivebox.core.middleware import ADMIN_LOGIN_HINT_COOKIE
|
||||
from archivebox.tests.conftest import ADMIN_TEST_HOST
|
||||
from archivebox.tests.test_archive_result_service import _run_shipped_snapshot_hook, _snapshot_hook_name
|
||||
|
||||
@ -858,11 +859,13 @@ class TestSnapshotOutputDeletion:
|
||||
|
||||
assert f'data-archive-result-ids="{result.id}"' in html
|
||||
assert 'title="Delete this output"' in html
|
||||
assert 'data-delete-handoff="1"' in html
|
||||
assert "const queuedOutputIds = new Set()" in html
|
||||
assert "action: 'delete_selected'" in html
|
||||
assert "window.location.assign(deleteUrl)" in html
|
||||
assert "/admin/core/archiveresult/" in html
|
||||
assert "[deleting]" in html
|
||||
assert ">×</button>" in html
|
||||
assert "delete-output-csrf" not in html
|
||||
assert "[data-archive-result-ids]:hover" in html
|
||||
assert "[data-archive-result-ids].delete-pending" in html
|
||||
assert "button.classList.toggle('delete-pending', queued)" in html
|
||||
@ -872,6 +875,55 @@ class TestSnapshotOutputDeletion:
|
||||
assert "data-archive-result-ids" not in anonymous_html
|
||||
assert 'title="Delete this output"' not in anonymous_html
|
||||
|
||||
request.COOKIES[ADMIN_LOGIN_HINT_COOKIE] = "1"
|
||||
hinted_html = SnapshotView.render_live_index(request, snapshot).content.decode()
|
||||
assert f'data-archive-result-ids="{result.id}"' in hinted_html
|
||||
assert "delete-output-csrf" not in hinted_html
|
||||
|
||||
def test_snapshot_delete_handoff_requires_superuser_confirmation_then_uses_standard_admin_action(self, client, snapshot, admin_user):
|
||||
from archivebox.core.models import ArchiveResult
|
||||
|
||||
result = self._create_output(snapshot)
|
||||
delete_url = reverse("admin:core_archiveresult_changelist")
|
||||
handoff_query = {
|
||||
"action": "delete_selected",
|
||||
ACTION_CHECKBOX_NAME: str(result.id),
|
||||
"snapshot": str(snapshot.id),
|
||||
}
|
||||
|
||||
logged_out = client.get(delete_url, handoff_query, HTTP_HOST=ADMIN_TEST_HOST)
|
||||
assert logged_out.status_code == 302
|
||||
assert ArchiveResult.objects.filter(pk=result.pk).exists()
|
||||
|
||||
staff_user = admin_user.__class__.objects.create_user(username="output-reviewer", password="testpassword", is_staff=True)
|
||||
client.force_login(staff_user)
|
||||
denied = client.get(delete_url, handoff_query, HTTP_HOST=ADMIN_TEST_HOST)
|
||||
assert denied.status_code == 403
|
||||
assert ArchiveResult.objects.filter(pk=result.pk).exists()
|
||||
|
||||
client.force_login(admin_user)
|
||||
confirmation = client.get(delete_url, handoff_query, HTTP_HOST=ADMIN_TEST_HOST)
|
||||
confirmation_html = confirmation.content.decode()
|
||||
assert confirmation.status_code == 200
|
||||
assert "Yes, I’m sure" in confirmation_html
|
||||
assert f'name="{ACTION_CHECKBOX_NAME}" value="{result.id}"' in confirmation_html
|
||||
assert confirmation["X-Frame-Options"] == "DENY"
|
||||
assert "frame-ancestors 'none'" in confirmation["Content-Security-Policy"]
|
||||
assert ArchiveResult.objects.filter(pk=result.pk).exists()
|
||||
|
||||
confirmed = client.post(
|
||||
f"{delete_url}?action=delete_selected&{ACTION_CHECKBOX_NAME}={result.id}&snapshot={snapshot.id}",
|
||||
{
|
||||
"action": "delete_selected",
|
||||
"post": "yes",
|
||||
ACTION_CHECKBOX_NAME: str(result.id),
|
||||
},
|
||||
HTTP_HOST=ADMIN_TEST_HOST,
|
||||
)
|
||||
assert confirmed.status_code == 302
|
||||
assert not ArchiveResult.objects.filter(pk=result.pk).exists()
|
||||
assert str(snapshot.id).replace("-", "")[-12:] in confirmed["Location"]
|
||||
|
||||
def test_batch_delete_removes_plugin_rows_files_and_refreshes_snapshot_size(self, client, snapshot, admin_user):
|
||||
from archivebox.core.models import ArchiveResult
|
||||
|
||||
|
||||
@ -385,6 +385,90 @@ class TestUrlRouting:
|
||||
assert (installed[0] / "sw.js").is_file()
|
||||
return installed[0]
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"mode",
|
||||
["auto", "safe-subdomains-fullreplay", "safe-onedomain-nojsreplay", "unsafe-onedomain-noadmin", "danger-onedomain-fullreplay"],
|
||||
)
|
||||
def test_snapshot_output_delete_handoff_is_non_mutating_in_every_security_mode(self, mode: str) -> None:
|
||||
self._run(
|
||||
"""
|
||||
ensure_admin_user()
|
||||
snapshot = get_snapshot()
|
||||
snapshot.config = {**snapshot.config, "PERMISSIONS": "public"}
|
||||
snapshot.save(update_fields=["config"])
|
||||
plugin = "security_delete_test"
|
||||
output_path = Path(snapshot.output_dir) / plugin / "output.txt"
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
output_path.write_text("security test", encoding="utf-8")
|
||||
result, _ = ArchiveResult.objects.update_or_create(
|
||||
snapshot=snapshot,
|
||||
plugin=plugin,
|
||||
defaults={
|
||||
"hook_name": "on_Snapshot__99_security_delete_test.py",
|
||||
"status": ArchiveResult.StatusChoices.SUCCEEDED,
|
||||
"output_str": "output.txt",
|
||||
"output_files": {"output.txt": {"size": 13, "mimetype": "text/plain"}},
|
||||
"output_size": 13,
|
||||
},
|
||||
)
|
||||
|
||||
client = Client(enforce_csrf_checks=True)
|
||||
assert client.login(username="testadmin", password="testpassword")
|
||||
admin_host = get_admin_host()
|
||||
admin_page = client.get("/admin/", HTTP_HOST=admin_host)
|
||||
if SERVER_CONFIG.CONTROL_PLANE_ENABLED:
|
||||
assert admin_page.status_code == 200
|
||||
else:
|
||||
assert admin_page.status_code == 403
|
||||
assert admin_page["X-Frame-Options"] == "DENY"
|
||||
assert "frame-ancestors 'none'" in admin_page["Content-Security-Policy"]
|
||||
|
||||
snapshot_host = get_snapshot_host(str(snapshot.id)) if SERVER_CONFIG.USES_SUBDOMAIN_ROUTING else get_base_host()
|
||||
snapshot_path = "/index.html" if SERVER_CONFIG.USES_SUBDOMAIN_ROUTING else f"/snapshot/{snapshot.id}/index.html"
|
||||
detail = client.get(snapshot_path, HTTP_HOST=snapshot_host)
|
||||
html = response_body(detail).decode("utf-8", "ignore")
|
||||
|
||||
assert detail.status_code == 200, (detail.status_code, detail.headers.get("Location"), html[:200])
|
||||
assert "delete-output-csrf" not in html
|
||||
if SERVER_CONFIG.CONTROL_PLANE_ENABLED:
|
||||
assert f'data-archive-result-ids="{result.id}"' in html, html[-1000:]
|
||||
assert 'data-delete-handoff="1"' in html, html[-1000:]
|
||||
else:
|
||||
assert f'data-archive-result-ids="{result.id}"' not in html
|
||||
|
||||
rejected = client.post(
|
||||
"/admin/core/archiveresult/",
|
||||
data={"action": "delete_selected", "post": "yes", "_selected_action": str(result.id)},
|
||||
HTTP_HOST=snapshot_host,
|
||||
)
|
||||
assert rejected.status_code == 403, (rejected.status_code, rejected.headers.get("Location"), response_body(rejected)[:200])
|
||||
assert ArchiveResult.objects.filter(pk=result.pk).exists()
|
||||
result.delete()
|
||||
print("OK")
|
||||
""",
|
||||
mode=mode,
|
||||
)
|
||||
|
||||
def test_cross_domain_admin_hint_only_marks_superusers(self) -> None:
|
||||
self._run(
|
||||
"""
|
||||
User = get_user_model()
|
||||
staff = User.objects.create_user(username="staff-hint-test", password="testpassword", is_staff=True)
|
||||
client = Client()
|
||||
client.force_login(staff)
|
||||
response = client.get("/admin/", HTTP_HOST=get_admin_host())
|
||||
assert response.status_code == 200
|
||||
assert client.cookies.get(ADMIN_LOGIN_HINT_COOKIE) is None or client.cookies[ADMIN_LOGIN_HINT_COOKIE].value != "1"
|
||||
|
||||
client.force_login(ensure_admin_user())
|
||||
response = client.get("/admin/", HTTP_HOST=get_admin_host())
|
||||
assert response.status_code == 200
|
||||
assert client.cookies[ADMIN_LOGIN_HINT_COOKIE].value == "1"
|
||||
print("OK")
|
||||
""",
|
||||
mode="safe-subdomains-fullreplay",
|
||||
)
|
||||
|
||||
def test_routes_util_and_web_public_redirect(self) -> None:
|
||||
self._run(
|
||||
"""
|
||||
|
||||
Loading…
Reference in New Issue
Block a user