fix: fall back when runuser fails in nextcloud

ariadne/services/nextcloud.py

@@ -105,23 +105,35 @@ class NextcloudService:
             settings.nextcloud_container,
         )
 
-    def _occ(self, args: list[str]) -> str:
+    def _exec_with_fallback(self, primary: list[str], fallback: list[str]) -> ExecResult:
-        command = ["runuser", "-u", "www-data", "--", "php", "/var/www/html/occ", *args]
+        try:
-        result = self._executor.exec(
+            return self._executor.exec(
-            command,
+                primary,
+                timeout_sec=settings.nextcloud_exec_timeout_sec,
+                check=True,
+            )
+        except ExecError as exc:
+            if "runuser: may not be used by non-root users" not in str(exc):
+                raise
+        return self._executor.exec(
+            fallback,
             timeout_sec=settings.nextcloud_exec_timeout_sec,
             check=True,
         )
+
+    def _occ(self, args: list[str]) -> str:
+        command = ["runuser", "-u", "www-data", "--", "php", "/var/www/html/occ", *args]
+        fallback = ["php", "/var/www/html/occ", *args]
+        result = self._exec_with_fallback(command, fallback)
         return result.stdout
 
     def run_cron(self) -> dict[str, Any]:
         if not settings.nextcloud_namespace:
             raise RuntimeError("nextcloud cron not configured")
         try:
-            self._executor.exec(
+            self._exec_with_fallback(
                 ["runuser", "-u", "www-data", "--", "php", "-f", "/var/www/html/cron.php"],
-                timeout_sec=settings.nextcloud_exec_timeout_sec,
+                ["php", "-f", "/var/www/html/cron.php"],
-                check=True,
             )
         except (ExecError, PodSelectionError, TimeoutError) as exc:
             return {"status": "error", "detail": str(exc)}

tests/test_nextcloud_sync.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import types
 
+from ariadne.k8s.exec import ExecError
 from ariadne.services import nextcloud as nextcloud_module
 from ariadne.services.nextcloud import NextcloudService, _parse_mail_export
 
@@ -182,6 +183,50 @@ def test_nextcloud_run_cron(monkeypatch) -> None:
     assert result["status"] == "ok"
 
 
+def test_nextcloud_occ_fallback(monkeypatch) -> None:
+    dummy_settings = types.SimpleNamespace(
+        nextcloud_namespace="nextcloud",
+        nextcloud_mail_sync_cronjob="nextcloud-mail-sync",
+        nextcloud_mail_sync_wait_timeout_sec=90.0,
+        nextcloud_mail_sync_job_ttl_sec=3600,
+        nextcloud_pod_label="app=nextcloud",
+        nextcloud_container="nextcloud",
+        nextcloud_exec_timeout_sec=30.0,
+        nextcloud_db_host="",
+        nextcloud_db_port=5432,
+        nextcloud_db_name="nextcloud",
+        nextcloud_db_user="nextcloud",
+        nextcloud_db_password="",
+        mailu_domain="bstein.dev",
+        mailu_host="mail.bstein.dev",
+    )
+    monkeypatch.setattr(nextcloud_module, "settings", dummy_settings)
+
+    svc = NextcloudService()
+
+    class DummyExec:
+        def __init__(self) -> None:
+            self.calls: list[list[str]] = []
+            self.fail = True
+
+        def exec(self, command, **_kwargs):
+            self.calls.append(command)
+            if self.fail:
+                self.fail = False
+                raise ExecError(
+                    "pod exec failed exit_code=1 stderr=runuser: may not be used by non-root users"
+                )
+            return types.SimpleNamespace(stdout="ok", stderr="", exit_code=0, ok=True)
+
+    executor = DummyExec()
+    svc._executor = executor
+
+    output = svc._occ(["status"])
+    assert output == "ok"
+    assert executor.calls[0][0] == "runuser"
+    assert executor.calls[1][0] == "php"
+
+
 def test_nextcloud_run_maintenance(monkeypatch) -> None:
     dummy_settings = types.SimpleNamespace(
         nextcloud_namespace="nextcloud",