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

(feature) Add staff updater page

Browse Source 
Add a Django updater view for checking configured Gitea releases, inspecting
the installed git checkout, fetching tags, pulling the current branch, and
running the configured native systemd update command.

Document the updater environment settings and keep the page staff-only so
readonly status users cannot trigger deployment actions.
This commit is contained in:
Peter van Arkel
2026-05-28 22:10:45 +02:00
parent b4833560b5
commit 0450f8bdb0
11 changed files with 516 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -275,6 +275,23 @@
.status.skipped { color: var(--muted); background: #f7f9fb; }
.stack { display: grid; gap: 5px; }
.stack.spaced { margin-bottom: 14px; }
.detail-list {
display: grid;
gap: 8px 14px;
grid-template-columns: minmax(120px, max-content) minmax(0, 1fr);
margin: 0 0 12px;
}
.detail-list dt {
color: var(--muted);
font-size: 12px;
font-weight: 750;
text-transform: uppercase;
}
.detail-list dd {
margin: 0;
min-width: 0;
overflow-wrap: anywhere;
}
.two-col { display: grid; gap: 18px; grid-template-columns: minmax(0, 1fr) minmax(0, 1fr); }
.panel-grid {
display: grid;
@@ -923,6 +940,7 @@
<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>
<a href="{% url 'updater' %}" {% if request.resolver_match.url_name == "updater" %}aria-current="page"{% endif %}>Updater</a>
{% endif %}
<a href="{% url 'purged_snapshots' %}" {% if request.resolver_match.url_name == "purged_snapshots" %}aria-current="page"{% endif %}>Purged</a>
</span>

122
src/pobsync_backend/templates/pobsync_backend/updater.html Normal file
View File

@@ -0,0 +1,122 @@
{% extends "pobsync_backend/base.html" %}
{% block title %}Updater | pobsync{% endblock %}
{% block content %}
<header class="page-header">
<div class="page-title">
<div class="page-kicker">Operations</div>
<h1>Updater</h1>
<div class="page-subtitle">Check Gitea releases, pull the installed git checkout, and run the native systemd updater.</div>
</div>
<section class="actions" aria-label="Updater actions">
<a class="button-link secondary" href="{% url 'dashboard' %}">Back to dashboard</a>
</section>
</header>
<section class="panel-grid">
<section class="panel">
<h2>Installed App</h2>
<dl class="detail-list">
<dt>Version</dt>
<dd>{{ status.installed_version }}</dd>
<dt>Git branch</dt>
<dd>{{ status.git.branch|default:"unknown" }}</dd>
<dt>Git commit</dt>
<dd>{{ status.git.commit|default:"unknown" }}</dd>
<dt>Git describe</dt>
<dd>{{ status.git.describe|default:"unknown" }}</dd>
<dt>App directory</dt>
<dd>{{ status.app_dir }}</dd>
</dl>
</section>
<section class="panel">
<h2>Release Check</h2>
<dl class="detail-list">
<dt>Status</dt>
<dd>
{% if status.update_available == True %}
<span class="status warning">update available</span>
{% elif status.update_available == False %}
<span class="status ok">up to date</span>
{% elif status.release_check_configured %}
<span class="status skipped">not checked</span>
{% else %}
<span class="status skipped">not configured</span>
{% endif %}
</dd>
<dt>Latest release</dt>
<dd>
{% if status.latest_release %}
{% if status.latest_release.html_url %}
<a href="{{ status.latest_release.html_url }}">
{% if status.latest_release.tag_name %}{{ status.latest_release.tag_name }}{% else %}{{ status.latest_release.name }}{% endif %}
</a>
{% else %}
{% if status.latest_release.tag_name %}{{ status.latest_release.tag_name }}{% else %}{{ status.latest_release.name }}{% endif %}
{% endif %}
{% else %}
none
{% endif %}
</dd>
<dt>Release endpoint</dt>
<dd>{% if status.release_check_configured %}configured{% else %}set POBSYNC_UPDATE_RELEASES_URL{% endif %}</dd>
</dl>
{% if status.release_error %}
<p class="status failed">{{ status.release_error }}</p>
{% endif %}
<form method="post" class="actions inline">
{% csrf_token %}
<button type="submit" name="action" value="check_release">Check releases</button>
</form>
</section>
</section>
<section class="panel">
<h2>Update Actions</h2>
<p class="muted">Run these from the installed checkout. The native updater may require a sudoers rule for the pobsync service user.</p>
<dl class="detail-list">
<dt>Git remote</dt>
<dd>{{ status.git_remote }}</dd>
<dt>Update command</dt>
<dd><code>{{ status.update_command }}</code></dd>
</dl>
<div class="actions">
<form method="post" class="inline-form">
{% csrf_token %}
<button class="secondary" type="submit" name="action" value="git_fetch">Fetch releases</button>
</form>
<form method="post" class="inline-form">
{% csrf_token %}
<button class="secondary" type="submit" name="action" value="git_pull">Pull current branch</button>
</form>
<form method="post" class="inline-form">
{% csrf_token %}
<button type="submit" name="action" value="run_update">Run native updater</button>
</form>
</div>
</section>
{% if action_result %}
<section class="panel">
<h2>Last Action Result</h2>
<dl class="detail-list">
<dt>Status</dt>
<dd><span class="status {% if action_result.ok %}ok{% else %}failed{% endif %}">{% if action_result.ok %}ok{% else %}failed{% endif %}</span></dd>
<dt>Exit code</dt>
<dd>{{ action_result.exit_code }}</dd>
<dt>Command</dt>
<dd><code>{{ action_result.command|join:" " }}</code></dd>
</dl>
{% if action_result.stdout %}
<h3>Stdout</h3>
<pre>{{ action_result.stdout }}</pre>
{% endif %}
{% if action_result.stderr %}
<h3>Stderr</h3>
<pre>{{ action_result.stderr }}</pre>
{% endif %}
</section>
{% endif %}
{% endblock %}

72
src/pobsync_backend/tests/test_updater.py Normal file
View File

@@ -0,0 +1,72 @@
from __future__ import annotations
import json
from unittest.mock import MagicMock, patch
from django.test import SimpleTestCase, override_settings
from pobsync_backend import updater
class UpdaterTests(SimpleTestCase):
@override_settings(
POBSYNC_UPDATE_RELEASES_URL="https://code.example.test/api/v1/repos/owner/pobsync/releases",
POBSYNC_UPDATE_RELEASES_TOKEN="secret",
)
def test_fetch_latest_release_reads_first_gitea_release(self) -> None:
response = MagicMock()
response.__enter__.return_value.read.return_value = json.dumps(
[
{
"tag_name": "v1.2.0",
"name": "1.2.0",
"html_url": "https://code.example.test/releases/v1.2.0",
}
]
).encode("utf-8")
with patch("pobsync_backend.updater.urlopen", return_value=response) as urlopen:
release = updater.fetch_latest_release()
self.assertEqual(release["tag_name"], "v1.2.0")
request = urlopen.call_args.args[0]
self.assertEqual(request.full_url, "https://code.example.test/api/v1/repos/owner/pobsync/releases")
self.assertEqual(request.headers["Authorization"], "token secret")
@override_settings(POBSYNC_UPDATE_RELEASES_URL="")
def test_collect_update_status_reports_unconfigured_release_check(self) -> None:
with patch("pobsync_backend.updater._git_status", return_value={"branch": "master"}):
status = updater.collect_update_status(check_release=True)
self.assertFalse(status["release_check_configured"])
self.assertEqual(status["release_error"], "POBSYNC_UPDATE_RELEASES_URL is not configured.")
self.assertIsNone(status["update_available"])
@override_settings(POBSYNC_UPDATE_GIT_REMOTE="upstream")
def test_run_git_fetch_uses_configured_remote(self) -> None:
completed = MagicMock(returncode=0, stdout="ok", stderr="")
with patch("pobsync_backend.updater.subprocess.run", return_value=completed) as run:
result = updater.run_git_fetch()
self.assertTrue(result.ok)
self.assertEqual(result.command, ["git", "fetch", "--tags", "--prune", "upstream"])
run.assert_called_once()
@override_settings(POBSYNC_UPDATE_GIT_REMOTE="origin")
def test_run_git_pull_rejects_detached_checkout(self) -> None:
with patch("pobsync_backend.updater._git_current_branch", return_value=""):
result = updater.run_git_pull()
self.assertFalse(result.ok)
self.assertEqual(result.exit_code, 2)
self.assertIn("not on a branch", result.stderr)
@override_settings(POBSYNC_UPDATE_COMMAND="sudo -n scripts/update-systemd --verbose")
def test_run_native_update_splits_configured_command(self) -> None:
completed = MagicMock(returncode=1, stdout="", stderr="sudo failed")
with patch("pobsync_backend.updater.subprocess.run", return_value=completed):
result = updater.run_native_update()
self.assertFalse(result.ok)
self.assertEqual(result.command, ["sudo", "-n", "scripts/update-systemd", "--verbose"])
self.assertEqual(result.stderr, "sudo failed")

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

@@ -62,6 +62,7 @@ class ViewTests(TestCase):
self.assertContains(response, reverse("ssh_credentials"))
self.assertContains(response, reverse("notification_targets"))
self.assertContains(response, reverse("logs"))
self.assertContains(response, reverse("updater"))
self.assertContains(response, reverse("purged_snapshots"))
self.assertContains(response, reverse("self_check"))
self.assertContains(response, reverse("changelog"))
@@ -82,6 +83,7 @@ class ViewTests(TestCase):
self.assertNotContains(response, reverse("ssh_credentials"))
self.assertNotContains(response, reverse("notification_targets"))
self.assertNotContains(response, reverse("logs"))
self.assertNotContains(response, reverse("updater"))
self.assertNotContains(response, reverse("self_check"))
self.assertNotContains(response, reverse("admin:index"))
@@ -135,6 +137,7 @@ class ViewTests(TestCase):
blocked_urls = [
reverse("ssh_credentials"),
reverse("logs"),
reverse("updater"),
reverse("self_check"),
reverse("edit_global_config"),
reverse("create_host_config"),
@@ -147,6 +150,90 @@ class ViewTests(TestCase):
response = self.client.get(url)
self.assertEqual(response.status_code, 403)
def test_updater_view_renders_status(self) -> None:
self.client.force_login(self.staff_user)
with patch("pobsync_backend.views.collect_update_status") as collect_update_status:
collect_update_status.return_value = {
"app_dir": "/opt/pobsync/app",
"installed_version": "1.1.0",
"release_check_configured": True,
"update_command": "sudo -n scripts/update-systemd",
"git_remote": "origin",
"git": {"branch": "master", "commit": "abc1234", "describe": "v1.1.0"},
"latest_release": None,
"release_error": "",
"update_available": None,
}
response = self.client.get(reverse("updater"))
self.assertEqual(response.status_code, 200)
self.assertContains(response, "Updater")
self.assertContains(response, "1.1.0")
self.assertContains(response, "master")
collect_update_status.assert_called_once_with(check_release=False)
def test_updater_check_release_requests_release_status(self) -> None:
self.client.force_login(self.staff_user)
with patch("pobsync_backend.views.collect_update_status") as collect_update_status:
collect_update_status.return_value = {
"app_dir": "/opt/pobsync/app",
"installed_version": "1.1.0",
"release_check_configured": True,
"update_command": "sudo -n scripts/update-systemd",
"git_remote": "origin",
"git": {"branch": "master", "commit": "abc1234", "describe": "v1.1.0"},
"latest_release": {"tag_name": "v1.2.0", "html_url": "https://code.example.test/releases/v1.2.0"},
"release_error": "",
"update_available": True,
}
response = self.client.post(reverse("updater"), {"action": "check_release"})
self.assertEqual(response.status_code, 200)
self.assertContains(response, "update available")
self.assertContains(response, "v1.2.0")
collect_update_status.assert_called_once_with(check_release=True)
def test_updater_git_fetch_runs_action_and_renders_output(self) -> None:
self.client.force_login(self.staff_user)
result = type(
"Result",
(),
{
"ok": True,
"exit_code": 0,
"command": ["git", "fetch", "--tags", "--prune", "origin"],
"stdout": "fetched",
"stderr": "",
},
)()
with patch("pobsync_backend.views.run_git_fetch", return_value=result) as run_git_fetch, patch(
"pobsync_backend.views.collect_update_status"
) as collect_update_status:
collect_update_status.return_value = {
"app_dir": "/opt/pobsync/app",
"installed_version": "1.1.0",
"release_check_configured": False,
"update_command": "sudo -n scripts/update-systemd",
"git_remote": "origin",
"git": {"branch": "master", "commit": "abc1234", "describe": "v1.1.0"},
"latest_release": None,
"release_error": "",
"update_available": None,
}
response = self.client.post(reverse("updater"), {"action": "git_fetch"})
self.assertEqual(response.status_code, 200)
self.assertContains(response, "fetched")
self.assertContains(response, "Updater action completed successfully.")
run_git_fetch.assert_called_once_with()
collect_update_status.assert_called_once_with(check_release=True)
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")

149
src/pobsync_backend/updater.py Normal file
View File

@@ -0,0 +1,149 @@
from __future__ import annotations
import json
import shlex
import subprocess
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
from django.conf import settings
from pobsync import __version__
@dataclass(frozen=True)
class CommandResult:
command: list[str]
exit_code: int
stdout: str
stderr: str
@property
def ok(self) -> bool:
return self.exit_code == 0
def collect_update_status(*, check_release: bool = False) -> dict[str, Any]:
app_dir = Path(settings.BASE_DIR)
status: dict[str, Any] = {
"app_dir": app_dir,
"installed_version": __version__,
"release_check_configured": bool(settings.POBSYNC_UPDATE_RELEASES_URL),
"update_command": settings.POBSYNC_UPDATE_COMMAND,
"git_remote": settings.POBSYNC_UPDATE_GIT_REMOTE,
"git": _git_status(app_dir),
"latest_release": None,
"release_error": "",
"update_available": None,
}
if check_release:
if not settings.POBSYNC_UPDATE_RELEASES_URL:
status["release_error"] = "POBSYNC_UPDATE_RELEASES_URL is not configured."
else:
try:
latest_release = fetch_latest_release()
status["latest_release"] = latest_release
status["update_available"] = _version_key(latest_release.get("tag_name", "")) != _version_key(__version__)
except (HTTPError, URLError, TimeoutError, json.JSONDecodeError, ValueError) as exc:
status["release_error"] = str(exc)
return status
def fetch_latest_release() -> dict[str, Any]:
request = Request(settings.POBSYNC_UPDATE_RELEASES_URL, headers={"Accept": "application/json"})
if settings.POBSYNC_UPDATE_RELEASES_TOKEN:
request.add_header("Authorization", f"token {settings.POBSYNC_UPDATE_RELEASES_TOKEN}")
with urlopen(request, timeout=10) as response:
payload = json.loads(response.read().decode("utf-8"))
if isinstance(payload, list):
if not payload:
raise ValueError("No releases were returned.")
release = payload[0]
elif isinstance(payload, dict):
release = payload
else:
raise ValueError("Release endpoint returned an unexpected payload.")
if not isinstance(release, dict):
raise ValueError("Release endpoint returned an unexpected release entry.")
return release
def run_git_fetch() -> CommandResult:
remote = settings.POBSYNC_UPDATE_GIT_REMOTE
return _run_command(["git", "fetch", "--tags", "--prune", remote])
def run_git_pull() -> CommandResult:
remote = settings.POBSYNC_UPDATE_GIT_REMOTE
branch = _git_current_branch(Path(settings.BASE_DIR))
if not branch:
return CommandResult(
command=["git", "pull", "--ff-only", remote],
exit_code=2,
stdout="",
stderr="Cannot pull automatically because the installed checkout is not on a branch.",
)
return _run_command(["git", "pull", "--ff-only", remote, branch])
def run_native_update() -> CommandResult:
return _run_command(shlex.split(settings.POBSYNC_UPDATE_COMMAND))
def _run_command(command: list[str]) -> CommandResult:
completed = subprocess.run(
command,
cwd=settings.BASE_DIR,
capture_output=True,
check=False,
text=True,
timeout=600,
)
return CommandResult(
command=command,
exit_code=completed.returncode,
stdout=completed.stdout[-6000:],
stderr=completed.stderr[-6000:],
)
def _git_status(app_dir: Path) -> dict[str, str]:
return {
"branch": _git_current_branch(app_dir),
"commit": _git_output(app_dir, ["git", "rev-parse", "--short", "HEAD"]),
"describe": _git_output(app_dir, ["git", "describe", "--tags", "--always", "--dirty"]),
}
def _git_current_branch(app_dir: Path) -> str:
branch = _git_output(app_dir, ["git", "branch", "--show-current"])
return branch or _git_output(app_dir, ["git", "rev-parse", "--abbrev-ref", "HEAD"])
def _git_output(app_dir: Path, command: list[str]) -> str:
try:
completed = subprocess.run(
command,
cwd=app_dir,
capture_output=True,
check=False,
text=True,
timeout=5,
)
except (FileNotFoundError, subprocess.TimeoutExpired):
return ""
if completed.returncode != 0:
return ""
return completed.stdout.strip()
def _version_key(value: str) -> str:
return value.strip().removeprefix("v")

35
src/pobsync_backend/views.py
View File

@@ -57,6 +57,7 @@ from .scheduler import next_due_after
from .snapshot_discovery import discover_snapshots, inspect_snapshot_discovery
from .ssh_keys import SshKeyError, delete_generated_key_files, generate_ssh_key, merge_known_hosts, scan_known_host
from .stats_summary import collect_dashboard_stats, collect_host_stats
from .updater import collect_update_status, run_git_fetch, run_git_pull, run_native_update
@status_view_required
@@ -333,6 +334,40 @@ def logs(request):
return render(request, "pobsync_backend/logs.html", context)
@control_panel_admin_required
def updater(request):
action_result = None
check_release = request.method == "POST" and request.POST.get("action") == "check_release"
if request.method == "POST":
action = request.POST.get("action")
if action == "git_fetch":
action_result = run_git_fetch()
check_release = True
elif action == "git_pull":
action_result = run_git_pull()
check_release = True
elif action == "run_update":
action_result = run_native_update()
check_release = True
elif action != "check_release":
messages.error(request, "Unknown updater action.")
if action_result is not None:
if action_result.ok:
messages.success(request, "Updater action completed successfully.")
else:
messages.error(request, f"Updater action failed with exit code {action_result.exit_code}.")
return render(
request,
"pobsync_backend/updater.html",
{
"status": collect_update_status(check_release=check_release),
"action_result": action_result,
},
)
@control_panel_admin_required
def notification_targets(request):
targets = NotificationTarget.objects.order_by("name")

4
src/pobsync_server/settings.py
View File

@@ -105,3 +105,7 @@ POBSYNC_BACKUP_ROOT = os.getenv("POBSYNC_BACKUP_ROOT", "/backups")
POBSYNC_ENV_FILE = os.getenv("POBSYNC_ENV_FILE", "/etc/pobsync/pobsync.env")
POBSYNC_SERVICE_USER = os.getenv("POBSYNC_SERVICE_USER", "pobsync")
POBSYNC_SERVICE_GROUP = os.getenv("POBSYNC_SERVICE_GROUP", "pobsync")
POBSYNC_UPDATE_RELEASES_URL = os.getenv("POBSYNC_UPDATE_RELEASES_URL", "")
POBSYNC_UPDATE_RELEASES_TOKEN = os.getenv("POBSYNC_UPDATE_RELEASES_TOKEN", "")
POBSYNC_UPDATE_GIT_REMOTE = os.getenv("POBSYNC_UPDATE_GIT_REMOTE", "origin")
POBSYNC_UPDATE_COMMAND = os.getenv("POBSYNC_UPDATE_COMMAND", "sudo -n scripts/update-systemd")

1
src/pobsync_server/urls.py
View File

@@ -13,6 +13,7 @@ urlpatterns = [
path("changelog/", views.changelog, name="changelog"),
path("self-check/", views.self_check, name="self_check"),
path("logs/", views.logs, name="logs"),
path("updater/", views.updater, name="updater"),
path("notifications/", views.notification_targets, name="notification_targets"),
path("notifications/new/", views.create_notification_target, name="create_notification_target"),
path("notifications/<int:target_id>/", views.edit_notification_target, name="edit_notification_target"),