hippogrief/pobsync
1
0
Fork 0
Code Issues 21 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

(feature) Add read-only access level to control panel

Browse Source 
Introduce a central access policy that lets authenticated non-staff users view
backup status pages while keeping credentials, logs, configs, and mutating
actions staff-only.

Hide sensitive navigation and host controls for read-only users, expose only
the status API to authenticated viewers, and document the two access levels.
This commit is contained in:
Peter van Arkel
2026-05-28 22:00:16 +02:00
parent 7f2bbe4d20
commit 81ee848f5f
13 changed files with 329 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
README.md
View File

@@ -131,6 +131,11 @@ Create a superuser if needed:
sudo -u pobsync pobsync-manage createsuperuser
```
The control panel supports two access levels. Django staff users can manage hosts, SSH keys, configs, retention,
notifications, logs, and administrative actions. Normal authenticated users can view backup status pages such as the
dashboard, hosts, runs, snapshots, schedules, purged history, changelog, and `/api/status/`, but cannot see SSH
credentials or run mutating actions.
For other Django management commands on native installs, use `pobsync-manage` so the production environment file is
loaded before Django starts:

39
src/pobsync_backend/access.py Normal file
View File

@@ -0,0 +1,39 @@
from __future__ import annotations
from collections.abc import Callable
from functools import wraps
from typing import Any
from django.contrib.auth.decorators import login_required
from django.core.exceptions import PermissionDenied
from django.http import HttpRequest, HttpResponse
def can_view_status(user) -> bool:
return bool(user.is_authenticated)
def can_manage_control_panel(user) -> bool:
return bool(user.is_authenticated and user.is_staff)
def status_view_required(view_func: Callable[..., HttpResponse]) -> Callable[..., HttpResponse]:
return login_required(view_func)
def control_panel_admin_required(view_func: Callable[..., HttpResponse]) -> Callable[..., HttpResponse]:
@login_required
@wraps(view_func)
def wrapper(request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse:
if not can_manage_control_panel(request.user):
raise PermissionDenied
return view_func(request, *args, **kwargs)
return wrapper
def access_context(request: HttpRequest) -> dict[str, Any]:
return {
"can_view_status": can_view_status(request.user),
"can_manage_control_panel": can_manage_control_panel(request.user),
}

12
src/pobsync_backend/api.py
View File

@@ -2,16 +2,16 @@ from __future__ import annotations
from typing import Any
from django.contrib.admin.views.decorators import staff_member_required
from django.db import connection
from django.db.models import Count
from django.http import JsonResponse
from django.utils import timezone
from .models import BackupRun, HostConfig, ScheduleConfig, SnapshotRecord
from .access import control_panel_admin_required, status_view_required
@staff_member_required
@control_panel_admin_required
def api_index(request) -> JsonResponse:
return JsonResponse(
{
@@ -26,7 +26,7 @@ def api_index(request) -> JsonResponse:
)
@staff_member_required
@status_view_required
def status(request) -> JsonResponse:
latest_run = BackupRun.objects.select_related("host", "snapshot").order_by("-created_at").first()
latest_schedule = ScheduleConfig.objects.select_related("host").order_by("-last_started_at", "-updated_at").first()
@@ -55,7 +55,7 @@ def status(request) -> JsonResponse:
)
@staff_member_required
@control_panel_admin_required
def hosts(request) -> JsonResponse:
host_qs = (
HostConfig.objects.annotate(snapshot_count=Count("snapshots", distinct=True), run_count=Count("runs", distinct=True))
@@ -65,7 +65,7 @@ def hosts(request) -> JsonResponse:
return JsonResponse({"ok": True, "hosts": [_host_payload(host) for host in host_qs]})
@staff_member_required
@control_panel_admin_required
def snapshots(request) -> JsonResponse:
snapshot_qs = SnapshotRecord.objects.select_related("host", "base").order_by("host__host", "-started_at", "dirname")
host_filter = request.GET.get("host")
@@ -78,7 +78,7 @@ def snapshots(request) -> JsonResponse:
return JsonResponse({"ok": True, "snapshots": [_snapshot_payload(snapshot) for snapshot in snapshot_qs[:limit]]})
@staff_member_required
@control_panel_admin_required
def runs(request) -> JsonResponse:
run_qs = BackupRun.objects.select_related("host", "snapshot").order_by("-created_at")
host_filter = request.GET.get("host")

9
src/pobsync_backend/context_processors.py Normal file
View File

@@ -0,0 +1,9 @@
from __future__ import annotations
from django.http import HttpRequest
from .access import access_context
def pobsync_access(request: HttpRequest) -> dict[str, object]:
return access_context(request)

16
src/pobsync_backend/templates/pobsync_backend/base.html
View File

@@ -919,17 +919,23 @@
<span class="nav-primary" aria-label="Primary navigation">
<a href="{% url 'dashboard' %}" {% if request.resolver_match.url_name == "dashboard" %}aria-current="page"{% endif %}>Dashboard</a>
<a href="{% url 'hosts_list' %}" {% if request.resolver_match.url_name == "hosts_list" or request.resolver_match.url_name == "host_detail" or request.resolver_match.url_name == "create_host_config" or request.resolver_match.url_name == "edit_host_config" or request.resolver_match.url_name == "edit_host_schedule" %}aria-current="page"{% endif %}>Hosts</a>
<a href="{% url 'ssh_credentials' %}" {% if request.resolver_match.url_name == "ssh_credentials" or request.resolver_match.url_name == "create_ssh_credential" or request.resolver_match.url_name == "generate_ssh_credential" or request.resolver_match.url_name == "edit_ssh_credential" %}aria-current="page"{% endif %}>SSH Keys</a>
<a href="{% url 'notification_targets' %}" {% if request.resolver_match.url_name == "notification_targets" or request.resolver_match.url_name == "create_notification_target" or request.resolver_match.url_name == "edit_notification_target" %}aria-current="page"{% endif %}>Notifications</a>
<a href="{% url 'logs' %}" {% if request.resolver_match.url_name == "logs" %}aria-current="page"{% endif %}>Logs</a>
{% if can_manage_control_panel %}
<a href="{% url 'ssh_credentials' %}" {% if request.resolver_match.url_name == "ssh_credentials" or request.resolver_match.url_name == "create_ssh_credential" or request.resolver_match.url_name == "generate_ssh_credential" or request.resolver_match.url_name == "edit_ssh_credential" %}aria-current="page"{% endif %}>SSH Keys</a>
<a href="{% url 'notification_targets' %}" {% if request.resolver_match.url_name == "notification_targets" or request.resolver_match.url_name == "create_notification_target" or request.resolver_match.url_name == "edit_notification_target" %}aria-current="page"{% endif %}>Notifications</a>
<a href="{% url 'logs' %}" {% if request.resolver_match.url_name == "logs" %}aria-current="page"{% endif %}>Logs</a>
{% endif %}
<a href="{% url 'purged_snapshots' %}" {% if request.resolver_match.url_name == "purged_snapshots" %}aria-current="page"{% endif %}>Purged</a>
</span>
<span class="spacer"></span>
<span class="nav-secondary" aria-label="System navigation">
<a href="{% url 'self_check' %}" {% if request.resolver_match.url_name == "self_check" %}aria-current="page"{% endif %}>Self Check</a>
{% if can_manage_control_panel %}
<a href="{% url 'self_check' %}" {% if request.resolver_match.url_name == "self_check" %}aria-current="page"{% endif %}>Self Check</a>
{% endif %}
<a href="{% url 'changelog' %}" {% if request.resolver_match.url_name == "changelog" %}aria-current="page"{% endif %}>Changelog</a>
<a href="/api/status/">Status API</a>
<a href="{% url 'admin:index' %}">Admin</a>
{% if can_manage_control_panel %}
<a href="{% url 'admin:index' %}">Admin</a>
{% endif %}
</span>
<span class="muted nav-user">{{ request.user.username }}</span>
</nav>

42
src/pobsync_backend/templates/pobsync_backend/dashboard.html
View File

@@ -9,27 +9,31 @@
<h1>Dashboard</h1>
<div class="page-subtitle">Backup health, required action, storage pressure, and recent activity in one place.</div>
</div>
<section class="actions" aria-label="Dashboard actions">
<a class="button-link" href="{% url 'create_host_config' %}">New host</a>
<a class="button-link secondary" href="{% url 'edit_global_config' %}">{% if global_config %}Edit global config{% else %}Create global config{% endif %}</a>
</section>
{% if can_manage_control_panel %}
<section class="actions" aria-label="Dashboard actions">
<a class="button-link" href="{% url 'create_host_config' %}">New host</a>
<a class="button-link secondary" href="{% url 'edit_global_config' %}">{% if global_config %}Edit global config{% else %}Create global config{% endif %}</a>
</section>
{% endif %}
</header>
{% if not global_config or not counts.hosts %}
<section class="panel">
<h2>Setup</h2>
{% if not global_config %}
<p class="muted">No default global config exists yet. Create it first so discovery, backups, and retention know where pobsync stores snapshots.</p>
<div class="actions inline">
<a class="button-link" href="{% url 'edit_global_config' %}">Create global config</a>
</div>
{% elif not counts.hosts %}
<p class="muted">Global config is ready. Add the first host to make this dashboard useful.</p>
<div class="actions inline">
<a class="button-link" href="{% url 'create_host_config' %}">Add first host</a>
</div>
{% endif %}
</section>
{% if can_manage_control_panel %}
{% if not global_config or not counts.hosts %}
<section class="panel">
<h2>Setup</h2>
{% if not global_config %}
<p class="muted">No default global config exists yet. Create it first so discovery, backups, and retention know where pobsync stores snapshots.</p>
<div class="actions inline">
<a class="button-link" href="{% url 'edit_global_config' %}">Create global config</a>
</div>
{% elif not counts.hosts %}
<p class="muted">Global config is ready. Add the first host to make this dashboard useful.</p>
<div class="actions inline">
<a class="button-link" href="{% url 'create_host_config' %}">Add first host</a>
</div>
{% endif %}
</section>
{% endif %}
{% endif %}
<div

91
src/pobsync_backend/templates/pobsync_backend/host_detail.html
View File

@@ -27,10 +27,12 @@
{{ retention_warning.incomplete_count }} incomplete snapshot(s) exist. Retention does not delete incomplete
snapshots automatically; inspect them before cleanup.
</div>
<form method="post" action="{% url 'resolve_host_incomplete_reviews' host.host %}">
{% csrf_token %}
<button type="submit" class="secondary">Mark incomplete reviewed</button>
</form>
{% if can_manage_control_panel %}
<form method="post" action="{% url 'resolve_host_incomplete_reviews' host.host %}">
{% csrf_token %}
<button type="submit" class="secondary">Mark incomplete reviewed</button>
</form>
{% endif %}
{% endif %}
{% if retention_warning.error %}
<div>{{ retention_warning.error }}</div>
@@ -80,8 +82,9 @@
</div>
</article>
<article class="panel host-control-panel">
<h2>Backup Control</h2>
{% if can_manage_control_panel %}
<article class="panel host-control-panel">
<h2>Backup Control</h2>
<div class="operator-state">
{% if active_run %}
<span class="status {{ active_run.status }}">{{ active_run.status }}</span>
@@ -121,10 +124,16 @@
<p class="muted">Real backups are blocked by failed preflight checks. Dry-runs may still be available when storage-only checks fail.</p>
{% endif %}
{% endif %}
</article>
</article>
{% endif %}
<article class="panel host-control-panel">
<h2>Schedule <a class="button-link secondary compact" href="{% url 'edit_host_schedule' host.host %}">Edit schedule</a></h2>
<h2>
Schedule
{% if can_manage_control_panel %}
<a class="button-link secondary compact" href="{% url 'edit_host_schedule' host.host %}">Edit schedule</a>
{% endif %}
</h2>
{% if schedule %}
<div class="host-control-meta">
<div><span class="label">Schedule expression</span><strong>{{ schedule.cron_expr }}</strong></div>
@@ -136,7 +145,9 @@
<p class="muted">Evaluated by the pobsync scheduler service.</p>
{% else %}
<p class="muted">No schedule configured.</p>
<a class="button-link secondary compact" href="{% url 'edit_host_schedule' host.host %}">Add schedule</a>
{% if can_manage_control_panel %}
<a class="button-link secondary compact" href="{% url 'edit_host_schedule' host.host %}">Add schedule</a>
{% endif %}
{% endif %}
</article>
@@ -252,8 +263,9 @@
</section>
{% endif %}
<section class="panel">
<h2>Host Check</h2>
{% if can_manage_control_panel %}
<section class="panel">
<h2>Host Check</h2>
<section class="grid" aria-label="Host check summary">
<div class="metric"><div class="label">OK</div><div class="value">{{ host_check_summary.ok }}</div></div>
<div class="metric"><div class="label">Warnings</div><div class="value">{{ host_check_summary.warning }}</div></div>
@@ -279,26 +291,32 @@
</article>
{% endfor %}
</div>
</section>
</section>
{% endif %}
<div class="panel-grid">
<section class="panel">
<h2>Configuration</h2>
<div class="host-control-meta">
<div><span class="label">Address</span><strong>{{ host.address }}</strong></div>
<div><span class="label">SSH key</span><strong>{{ host.ssh_credential|default:"global default" }}</strong></div>
<div><span class="label">SSH</span><strong>{{ host.ssh_user|default:"global" }}{% if host.ssh_port %}:{{ host.ssh_port }}{% endif %}</strong></div>
{% if can_manage_control_panel %}
<div><span class="label">SSH key</span><strong>{{ host.ssh_credential|default:"global default" }}</strong></div>
<div><span class="label">SSH</span><strong>{{ host.ssh_user|default:"global" }}{% if host.ssh_port %}:{{ host.ssh_port }}{% endif %}</strong></div>
{% endif %}
<div><span class="label">Backup source</span><strong>{{ host.source_root|default:"global default" }}</strong></div>
<div><span class="label">Retention</span><strong>d{{ host.retention_daily }} w{{ host.retention_weekly }} m{{ host.retention_monthly }} y{{ host.retention_yearly }}</strong></div>
</div>
<div class="actions inline">
<a class="button-link secondary compact" href="{% url 'edit_host_config' host.host %}">Edit config</a>
<a class="button-link secondary compact" href="{% url 'host_retention_plan' host.host %}">Plan retention</a>
</div>
{% if can_manage_control_panel %}
<div class="actions inline">
<a class="button-link secondary compact" href="{% url 'edit_host_config' host.host %}">Edit config</a>
<a class="button-link secondary compact" href="{% url 'host_retention_plan' host.host %}">Plan retention</a>
</div>
{% endif %}
</section>
<section class="panel">
<h2>Connection Preflight &amp; SSH</h2>
{% if can_manage_control_panel %}
<section class="panel">
<h2>Connection Preflight &amp; SSH</h2>
{% if last_preflight %}
<div class="host-control-meta">
<div>
@@ -341,7 +359,8 @@
{% endfor %}
</div>
{% endif %}
</section>
</section>
{% endif %}
<section class="panel">
<h2>Snapshot Storage</h2>
@@ -360,16 +379,18 @@
</div>
{% endif %}
</div>
<div class="actions inline">
<form method="post" action="{% url 'discover_host_snapshots' host.host %}">
{% csrf_token %}
<button type="submit" class="secondary compact">Discover snapshots</button>
</form>
<form method="post" action="{% url 'prepare_host_directories' host.host %}">
{% csrf_token %}
<button type="submit" class="secondary compact">Prepare directories</button>
</form>
</div>
{% if can_manage_control_panel %}
<div class="actions inline">
<form method="post" action="{% url 'discover_host_snapshots' host.host %}">
{% csrf_token %}
<button type="submit" class="secondary compact">Discover snapshots</button>
</form>
<form method="post" action="{% url 'prepare_host_directories' host.host %}">
{% csrf_token %}
<button type="submit" class="secondary compact">Prepare directories</button>
</form>
</div>
{% endif %}
</section>
</div>
@@ -451,8 +472,9 @@
</section>
{% endif %}
<section class="panel">
<h2>Backup Options</h2>
{% if can_manage_control_panel %}
<section class="panel">
<h2>Backup Options</h2>
<p class="muted">Use this when the quick actions above need a custom label, include/exclude override, or prune limit.</p>
<form method="post" action="{% url 'queue_manual_backup' host.host %}" class="form-grid">
{% csrf_token %}
@@ -471,7 +493,8 @@
<button type="submit" {% if not can_queue_dry_run and not can_queue_real_backup %}disabled{% endif %}>Queue with options</button>
</div>
</form>
</section>
</section>
{% endif %}
<section class="panel">
<h2>Latest Runs <a class="button-link secondary compact" href="{% url 'runs_list' %}?host={{ host.host }}">View all</a></h2>

8
src/pobsync_backend/templates/pobsync_backend/hosts_list.html
View File

@@ -9,9 +9,11 @@
<h1>Hosts</h1>
<div class="page-subtitle">Configured backup targets, schedules, retention state, and host-level controls.</div>
</div>
<section class="actions" aria-label="Host actions">
<a class="button-link" href="{% url 'create_host_config' %}">New host</a>
</section>
{% if can_manage_control_panel %}
<section class="actions" aria-label="Host actions">
<a class="button-link" href="{% url 'create_host_config' %}">New host</a>
</section>
{% endif %}
</header>
<section class="grid dashboard-summary-grid" aria-label="Host summary">

10
src/pobsync_backend/templates/pobsync_backend/partials/run_detail_live.html
View File

@@ -32,7 +32,7 @@
</section>
{% endif %}
{% if run.status == "failed" or run.status == "warning" %}
{% if can_manage_control_panel and run.status == "failed" or can_manage_control_panel and run.status == "warning" %}
{% if not run.reviewed_at %}
<section class="panel highlight warning">
<h2>Review Required</h2>
@@ -78,7 +78,7 @@
{% endif %}
<div>
<strong>Log:</strong>
{% if dry_run_summary.log_available %}
{% if dry_run_summary.log_available and can_manage_control_panel %}
<a href="{% url 'run_rsync_log' run.id %}">Open full rsync log</a>
{% elif rsync_log_path %}
<span class="muted">{{ rsync_log_path }} (missing)</span>
@@ -155,7 +155,7 @@
{% if live_progress.log.path %}
<div>
<strong>Log:</strong>
{% if live_progress.log.exists %}
{% if live_progress.log.exists and can_manage_control_panel %}
<a href="{% url 'run_rsync_log' run.id %}">Open full rsync log</a>
{% else %}
<span class="muted">{{ live_progress.log.path }} (missing)</span>
@@ -189,7 +189,7 @@
<div><strong>Base:</strong> {{ run.base_path|default:"" }}</div>
<div>
<strong>Rsync log:</strong>
{% if rsync_log_exists %}
{% if rsync_log_exists and can_manage_control_panel %}
<a href="{% url 'run_rsync_log' run.id %}">{{ rsync_log_path }}</a>
{% elif rsync_log_path %}
<span class="muted">{{ rsync_log_path }} (missing)</span>
@@ -204,7 +204,7 @@
<section class="panel">
<h2>Rsync Log</h2>
<div class="stack spaced">
{% if rsync_log_exists %}
{% if rsync_log_exists and can_manage_control_panel %}
<div><a href="{% url 'run_rsync_log' run.id %}">Open full rsync log</a></div>
<div class="muted">{{ rsync_log_path }}</div>
{% elif rsync_log_path %}

15
src/pobsync_backend/tests/test_api.py
View File

@@ -18,6 +18,12 @@ class ApiTests(TestCase):
is_staff=True,
is_superuser=True,
)
self.readonly_user = user_model.objects.create_user(
username="viewer",
password="secret",
is_staff=False,
is_superuser=False,
)
def test_api_requires_staff_login(self) -> None:
response = self.client.get("/api/hosts/")
@@ -25,6 +31,15 @@ class ApiTests(TestCase):
self.assertEqual(response.status_code, 302)
self.assertIn("/admin/login/", response["Location"])
def test_readonly_user_can_access_status_endpoint_only(self) -> None:
self.client.force_login(self.readonly_user)
status_response = self.client.get("/api/status/")
hosts_response = self.client.get("/api/hosts/")
self.assertEqual(status_response.status_code, 200)
self.assertEqual(hosts_response.status_code, 403)
def test_hosts_endpoint_returns_counts_and_schedule(self) -> None:
self.client.force_login(self.staff_user)
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")

93
src/pobsync_backend/tests/test_views.py
View File

@@ -36,6 +36,12 @@ class ViewTests(TestCase):
is_staff=True,
is_superuser=True,
)
self.readonly_user = user_model.objects.create_user(
username="viewer",
password="secret",
is_staff=False,
is_superuser=False,
)
def test_dashboard_requires_staff_login(self) -> None:
response = self.client.get(reverse("dashboard"))
@@ -63,6 +69,22 @@ class ViewTests(TestCase):
self.assertContains(response, reverse("admin:index"))
self.assertContains(response, '<a href="/" aria-current="page">Dashboard</a>', html=False)
def test_readonly_navigation_hides_admin_and_sensitive_links(self) -> None:
self.client.force_login(self.readonly_user)
response = self.client.get(reverse("dashboard"))
self.assertEqual(response.status_code, 200)
self.assertContains(response, reverse("dashboard"))
self.assertContains(response, reverse("hosts_list"))
self.assertContains(response, reverse("changelog"))
self.assertContains(response, "/api/status/")
self.assertNotContains(response, reverse("ssh_credentials"))
self.assertNotContains(response, reverse("notification_targets"))
self.assertNotContains(response, reverse("logs"))
self.assertNotContains(response, reverse("self_check"))
self.assertNotContains(response, reverse("admin:index"))
def test_base_navigation_marks_current_secondary_page(self) -> None:
self.client.force_login(self.staff_user)
@@ -71,12 +93,81 @@ class ViewTests(TestCase):
self.assertEqual(response.status_code, 200)
self.assertContains(response, f'<a href="{reverse("self_check")}" aria-current="page">Self Check</a>', html=False)
def test_changelog_requires_staff_login(self) -> None:
def test_changelog_requires_login(self) -> None:
response = self.client.get(reverse("changelog"))
self.assertEqual(response.status_code, 302)
self.assertIn("/admin/login/", response["Location"])
def test_readonly_user_can_view_status_pages(self) -> None:
self.client.force_login(self.readonly_user)
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
snapshot = SnapshotRecord.objects.create(
host=host,
kind=SnapshotRecord.Kind.SCHEDULED,
dirname="20260519-021500Z__ABCDEFGH",
path="/backups/web-01/scheduled/20260519-021500Z__ABCDEFGH",
status="success",
)
run = BackupRun.objects.create(host=host, status=BackupRun.Status.SUCCESS, snapshot=snapshot)
urls = [
reverse("dashboard"),
reverse("hosts_list"),
reverse("host_detail", args=[host.host]),
reverse("runs_list"),
reverse("run_detail", args=[run.id]),
reverse("snapshots_list"),
reverse("snapshot_detail", args=[snapshot.id]),
reverse("schedules_list"),
reverse("purged_snapshots"),
]
for url in urls:
with self.subTest(url=url):
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
def test_readonly_user_cannot_access_sensitive_or_mutating_views(self) -> None:
self.client.force_login(self.readonly_user)
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
blocked_urls = [
reverse("ssh_credentials"),
reverse("logs"),
reverse("self_check"),
reverse("edit_global_config"),
reverse("create_host_config"),
reverse("edit_host_config", args=[host.host]),
reverse("edit_host_schedule", args=[host.host]),
]
for url in blocked_urls:
with self.subTest(url=url):
response = self.client.get(url)
self.assertEqual(response.status_code, 403)
def test_readonly_host_detail_hides_backup_controls_and_sensitive_config(self) -> None:
self.client.force_login(self.readonly_user)
GlobalConfig.objects.create(name="default", backup_root="/backups")
credential = SshCredential.objects.create(name="root-key", key_path="/var/lib/pobsync/state/root-key")
host = HostConfig.objects.create(
host="web-01",
address="web-01.example.test",
ssh_credential=credential,
ssh_user="root",
)
response = self.client.get(reverse("host_detail", args=[host.host]))
self.assertEqual(response.status_code, 200)
self.assertContains(response, "Host Status")
self.assertNotContains(response, "Backup Control")
self.assertNotContains(response, "Backup Options")
self.assertNotContains(response, "Connection Preflight")
self.assertNotContains(response, "root-key")
self.assertNotContains(response, reverse("queue_manual_backup", args=[host.host]))
def test_changelog_renders_repository_changelog(self) -> None:
self.client.force_login(self.staff_user)
with TemporaryDirectory() as tmp:

112
src/pobsync_backend/views.py
View File

@@ -10,7 +10,6 @@ from pathlib import Path
from urllib.parse import urlencode
from django.contrib import messages
from django.contrib.admin.views.decorators import staff_member_required
from django.conf import settings
from django.http import FileResponse, Http404
from django.db.models import Count, Q
@@ -24,6 +23,7 @@ from pobsync import __version__
from pobsync.errors import PobsyncError
from pobsync.config.defaults import DEFAULT_EXCLUDES, DEFAULT_RSYNC_ARGS
from .access import access_context, control_panel_admin_required, status_view_required
from .backup_runner import queue_backup_run
from .config_checks import collect_effective_host_config_checks, collect_global_config_checks
from .forms import (
@@ -59,19 +59,19 @@ from .ssh_keys import SshKeyError, delete_generated_key_files, generate_ssh_key,
from .stats_summary import collect_dashboard_stats, collect_host_stats
@staff_member_required
@status_view_required
def dashboard(request):
return render(request, "pobsync_backend/dashboard.html", _dashboard_context())
return render(request, "pobsync_backend/dashboard.html", _dashboard_context(request))
@staff_member_required
@status_view_required
def dashboard_priority_live(request):
return render(request, "pobsync_backend/partials/dashboard_priority.html", _dashboard_context())
return render(request, "pobsync_backend/partials/dashboard_priority.html", _dashboard_context(request))
@staff_member_required
@status_view_required
def dashboard_hosts_live(request):
return render(request, "pobsync_backend/partials/dashboard_hosts.html", _dashboard_context())
return render(request, "pobsync_backend/partials/dashboard_hosts.html", _dashboard_context(request))
def _host_cards_context(*, enabled: str = "") -> dict[str, object]:
@@ -120,7 +120,7 @@ def _host_cards_context(*, enabled: str = "") -> dict[str, object]:
}
def _dashboard_context() -> dict[str, object]:
def _dashboard_context(request) -> dict[str, object]:
global_config = GlobalConfig.objects.filter(name="default").first()
host_context = _host_cards_context()
hosts = host_context["hosts"]
@@ -129,6 +129,7 @@ def _dashboard_context() -> dict[str, object]:
recent_runs = BackupRun.objects.select_related("host", "snapshot").order_by("-created_at", "-id")[:6]
stats_summary = collect_dashboard_stats(hosts=hosts, global_config=global_config)
context = {
**access_context(request),
"hosts": hosts,
"global_config": global_config,
"stats_summary": stats_summary,
@@ -159,7 +160,7 @@ def _dashboard_context() -> dict[str, object]:
return context
@staff_member_required
@status_view_required
def hosts_list(request):
enabled = request.GET.get("enabled", "").strip()
if enabled not in {"", "yes", "no"}:
@@ -171,15 +172,16 @@ def hosts_list(request):
request,
"pobsync_backend/hosts_list.html",
{
**access_context(request),
**context,
"global_config": global_config,
"show_host_controls": True,
"show_host_controls": request.user.is_staff,
"total_count": HostConfig.objects.count(),
},
)
@staff_member_required
@control_panel_admin_required
@require_POST
def update_host_state(request, host: str):
host_config = get_object_or_404(HostConfig, host=host)
@@ -290,7 +292,7 @@ def _retention_warning_summary(retention_warning) -> str:
return " ".join(parts)
@staff_member_required
@status_view_required
def changelog(request):
changelog_path = Path(settings.BASE_DIR) / "CHANGELOG.md"
try:
@@ -312,7 +314,7 @@ def changelog(request):
)
@staff_member_required
@control_panel_admin_required
def self_check(request):
checks = collect_self_checks()
return render(
@@ -325,13 +327,13 @@ def self_check(request):
)
@staff_member_required
@control_panel_admin_required
def logs(request):
context = _log_context(request)
return render(request, "pobsync_backend/logs.html", context)
@staff_member_required
@control_panel_admin_required
def notification_targets(request):
targets = NotificationTarget.objects.order_by("name")
deliveries = NotificationDelivery.objects.select_related("target", "run", "run__host").order_by("-created_at")[:12]
@@ -345,7 +347,7 @@ def notification_targets(request):
)
@staff_member_required
@control_panel_admin_required
def create_notification_target(request):
if request.method == "POST":
form = NotificationTargetForm(request.POST)
@@ -367,7 +369,7 @@ def create_notification_target(request):
)
@staff_member_required
@control_panel_admin_required
def edit_notification_target(request, target_id: int):
target = get_object_or_404(NotificationTarget, id=target_id)
if request.method == "POST":
@@ -390,7 +392,7 @@ def edit_notification_target(request, target_id: int):
)
@staff_member_required
@status_view_required
def runs_list(request):
status = request.GET.get("status", "").strip()
run_type = request.GET.get("type", "").strip()
@@ -422,7 +424,7 @@ def runs_list(request):
return render(request, "pobsync_backend/runs_list.html", context)
@staff_member_required
@status_view_required
def snapshots_list(request):
kind = request.GET.get("kind", "").strip()
status = request.GET.get("status", "").strip()
@@ -448,7 +450,7 @@ def snapshots_list(request):
return render(request, "pobsync_backend/snapshots_list.html", context)
@staff_member_required
@status_view_required
def schedules_list(request):
enabled = request.GET.get("enabled", "").strip()
prune = request.GET.get("prune", "").strip()
@@ -486,7 +488,7 @@ def schedules_list(request):
return render(request, "pobsync_backend/schedules_list.html", context)
@staff_member_required
@status_view_required
def purged_snapshots(request):
host = request.GET.get("host", "").strip()
action = request.GET.get("action", "").strip()
@@ -507,7 +509,7 @@ def purged_snapshots(request):
return render(request, "pobsync_backend/purged_snapshots.html", context)
@staff_member_required
@control_panel_admin_required
def ssh_credentials(request):
context = {
"credentials": SshCredential.objects.order_by("name"),
@@ -515,7 +517,7 @@ def ssh_credentials(request):
return render(request, "pobsync_backend/ssh_credentials.html", context)
@staff_member_required
@control_panel_admin_required
def create_ssh_credential(request):
if request.method == "POST":
form = SshCredentialForm(request.POST, request.FILES)
@@ -536,7 +538,7 @@ def create_ssh_credential(request):
)
@staff_member_required
@control_panel_admin_required
def generate_ssh_credential(request):
if request.method == "POST":
form = SshCredentialGenerateForm(request.POST)
@@ -572,7 +574,7 @@ def generate_ssh_credential(request):
)
@staff_member_required
@control_panel_admin_required
def edit_ssh_credential(request, credential_id: int):
credential = get_object_or_404(SshCredential, id=credential_id)
if request.method == "POST":
@@ -594,7 +596,7 @@ def edit_ssh_credential(request, credential_id: int):
)
@staff_member_required
@control_panel_admin_required
@require_POST
def delete_ssh_credential(request, credential_id: int):
credential = get_object_or_404(SshCredential, id=credential_id)
@@ -618,7 +620,7 @@ def delete_ssh_credential(request, credential_id: int):
return redirect("ssh_credentials")
@staff_member_required
@control_panel_admin_required
def edit_global_config(request):
global_config = GlobalConfig.objects.filter(name="default").first()
if request.method == "POST":
@@ -644,7 +646,7 @@ def edit_global_config(request):
)
@staff_member_required
@control_panel_admin_required
def create_host_config(request):
if request.method == "POST":
form = CreateHostConfigForm(request.POST)
@@ -672,7 +674,7 @@ def create_host_config(request):
)
@staff_member_required
@status_view_required
def host_detail(request, host: str):
host_config = get_object_or_404(HostConfig, host=host)
global_config = GlobalConfig.objects.filter(name="default").first()
@@ -685,7 +687,9 @@ def host_detail(request, host: str):
has_global_config = global_config is not None
backup_gate = collect_backup_gate(host_config, global_config)
stats_summary = collect_host_stats(host=host_config, limit=10)
can_manage = request.user.is_staff
context = {
**access_context(request),
"host": host_config,
"schedule": schedule,
"retention_warning": _retention_warning_for_host(host_config, schedule),
@@ -696,11 +700,11 @@ def host_detail(request, host: str):
"host_check_summary": summarize_self_checks(backup_gate.checks),
"backup_gate": backup_gate,
"last_preflight": (host_config.config or {}).get("last_preflight") if isinstance(host_config.config, dict) else {},
"effective_config": effective_host_config_preview(host_config, global_config) if global_config else {},
"effective_config": effective_host_config_preview(host_config, global_config) if global_config and can_manage else {},
"stats_summary": stats_summary,
"manual_backup_form": ManualBackupForm(initial=_default_manual_backup_initial(host_config)),
"can_queue_dry_run": host_config.enabled and has_global_config and backup_gate.can_queue_dry_run and active_run is None,
"can_queue_real_backup": host_config.enabled and has_global_config and backup_gate.can_queue_real and active_run is None,
"can_queue_dry_run": can_manage and host_config.enabled and has_global_config and backup_gate.can_queue_dry_run and active_run is None,
"can_queue_real_backup": can_manage and host_config.enabled and has_global_config and backup_gate.can_queue_real and active_run is None,
"has_global_config": has_global_config,
"active_run": active_run,
"latest_runs": host_config.runs.select_related("snapshot").order_by("-created_at")[:10],
@@ -720,7 +724,7 @@ def host_detail(request, host: str):
return render(request, "pobsync_backend/host_detail.html", context)
@staff_member_required
@control_panel_admin_required
@require_POST
def prepare_host_directories(request, host: str):
host_config = get_object_or_404(HostConfig, host=host)
@@ -733,7 +737,7 @@ def prepare_host_directories(request, host: str):
return redirect("host_detail", host=host_config.host)
@staff_member_required
@control_panel_admin_required
@require_POST
def scan_host_known_key(request, host: str):
host_config = get_object_or_404(HostConfig, host=host)
@@ -755,7 +759,7 @@ def scan_host_known_key(request, host: str):
return redirect("host_detail", host=host_config.host)
@staff_member_required
@control_panel_admin_required
@require_POST
def run_host_preflight(request, host: str):
host_config = get_object_or_404(HostConfig, host=host)
@@ -783,7 +787,7 @@ def run_host_preflight(request, host: str):
return redirect("host_detail", host=host_config.host)
@staff_member_required
@control_panel_admin_required
@require_POST
def queue_manual_backup(request, host: str):
host_config = get_object_or_404(HostConfig, host=host)
@@ -823,22 +827,22 @@ def queue_manual_backup(request, host: str):
return redirect("run_detail", run_id=run.id)
@staff_member_required
@status_view_required
def run_detail(request, run_id: int):
run = get_object_or_404(BackupRun.objects.select_related("host", "snapshot"), id=run_id)
return render(request, "pobsync_backend/run_detail.html", _run_detail_context(run))
return render(request, "pobsync_backend/run_detail.html", _run_detail_context(run, request=request))
@staff_member_required
@status_view_required
def run_detail_live(request, run_id: int):
run = get_object_or_404(BackupRun.objects.select_related("host", "snapshot"), id=run_id)
context = _run_detail_context(run)
context = _run_detail_context(run, request=request)
response = render(request, "pobsync_backend/partials/run_detail_live.html", context)
response["X-Pobsync-Refresh-Active"] = "true" if context["can_auto_refresh"] else "false"
return response
def _run_detail_context(run: BackupRun) -> dict[str, object]:
def _run_detail_context(run: BackupRun, *, request=None) -> dict[str, object]:
result = run.result if isinstance(run.result, dict) else {}
run_stats = result.get("stats") if isinstance(result.get("stats"), dict) else {}
rsync_result = result.get("rsync") if isinstance(result.get("rsync"), dict) else {}
@@ -848,8 +852,10 @@ def _run_detail_context(run: BackupRun) -> dict[str, object]:
rsync_log_path = _run_rsync_log_path(run)
rsync_log_tail = _run_rsync_log_tail(rsync_result, rsync_log_path)
requested = result.get("requested") if isinstance(result.get("requested"), dict) else {}
can_cancel = run.status in {BackupRun.Status.QUEUED, BackupRun.Status.RUNNING}
can_manage = bool(request and request.user.is_staff)
can_cancel = can_manage and run.status in {BackupRun.Status.QUEUED, BackupRun.Status.RUNNING}
return {
**(access_context(request) if request is not None else {}),
"run": run,
"can_cancel": can_cancel,
"can_auto_refresh": can_cancel,
@@ -880,7 +886,7 @@ def _run_detail_context(run: BackupRun) -> dict[str, object]:
}
@staff_member_required
@control_panel_admin_required
def run_rsync_log(request, run_id: int):
run = get_object_or_404(BackupRun.objects.select_related("host", "snapshot"), id=run_id)
log_path = _run_rsync_log_path(run)
@@ -889,7 +895,7 @@ def run_rsync_log(request, run_id: int):
return FileResponse(log_path.open("rb"), content_type="text/plain; charset=utf-8")
@staff_member_required
@control_panel_admin_required
@require_POST
def cancel_run(request, run_id: int):
run = get_object_or_404(BackupRun.objects.select_related("host"), id=run_id)
@@ -913,7 +919,7 @@ def cancel_run(request, run_id: int):
return redirect("run_detail", run_id=run.id)
@staff_member_required
@control_panel_admin_required
@require_POST
def resolve_run_review(request, run_id: int):
run = get_object_or_404(BackupRun.objects.select_related("host"), id=run_id)
@@ -931,7 +937,7 @@ def resolve_run_review(request, run_id: int):
return _redirect_after_run_review(request, run)
@staff_member_required
@control_panel_admin_required
@require_POST
def resolve_host_incomplete_reviews(request, host: str):
host_config = get_object_or_404(HostConfig, host=host)
@@ -947,7 +953,7 @@ def resolve_host_incomplete_reviews(request, host: str):
return redirect("host_detail", host=host_config.host)
@staff_member_required
@status_view_required
def snapshot_detail(request, snapshot_id: int):
snapshot = get_object_or_404(
SnapshotRecord.objects.select_related("host", "base").prefetch_related("derived_snapshots", "backup_runs"),
@@ -965,7 +971,7 @@ def snapshot_detail(request, snapshot_id: int):
return render(request, "pobsync_backend/snapshot_detail.html", context)
@staff_member_required
@control_panel_admin_required
@require_POST
def discover_host_snapshots(request, host: str):
host_config = get_object_or_404(HostConfig, host=host)
@@ -986,7 +992,7 @@ def discover_host_snapshots(request, host: str):
return redirect("host_detail", host=host_config.host)
@staff_member_required
@control_panel_admin_required
def host_retention_plan(request, host: str):
host_config = get_object_or_404(HostConfig, host=host)
kind = request.GET.get("kind", "scheduled")
@@ -1037,7 +1043,7 @@ def host_retention_plan(request, host: str):
return render(request, "pobsync_backend/retention_plan.html", context)
@staff_member_required
@control_panel_admin_required
@require_POST
def apply_host_retention(request, host: str):
host_config = get_object_or_404(HostConfig, host=host)
@@ -1091,7 +1097,7 @@ def apply_host_retention(request, host: str):
return target
@staff_member_required
@control_panel_admin_required
@require_POST
def cleanup_host_incomplete_snapshots(request, host: str):
host_config = get_object_or_404(HostConfig, host=host)
@@ -1126,7 +1132,7 @@ def cleanup_host_incomplete_snapshots(request, host: str):
return redirect("host_retention_plan", host=host_config.host)
@staff_member_required
@control_panel_admin_required
def edit_host_config(request, host: str):
host_config = get_object_or_404(HostConfig, host=host)
global_config = GlobalConfig.objects.filter(name="default").first()
@@ -1152,7 +1158,7 @@ def edit_host_config(request, host: str):
)
@staff_member_required
@control_panel_admin_required
def edit_host_schedule(request, host: str):
host_config = get_object_or_404(HostConfig, host=host)
schedule = _schedule_for_host(host_config)

3
src/pobsync_server/settings.py
View File

@@ -48,6 +48,7 @@ TEMPLATES = [
"django.template.context_processors.request",
"django.contrib.auth.context_processors.auth",
"django.contrib.messages.context_processors.messages",
"pobsync_backend.context_processors.pobsync_access",
],
},
},
@@ -55,6 +56,8 @@ TEMPLATES = [
WSGI_APPLICATION = "pobsync_server.wsgi.application"
LOGIN_URL = "/admin/login/"
def _database_config() -> dict[str, object]:
engine = os.getenv("POBSYNC_DB_ENGINE", "sqlite").strip().lower()