NAS-132001 / 25.04 / Fix __job_by_credential_and_id crashing for non-full-admins and internally ran jobs

src/middlewared/middlewared/service/core_service.py

@@ -101,6 +101,11 @@ def __job_by_credential_and_id(self, credential, job_id):
 
         job = self.middleware.jobs[job_id]
 
+        if job.credentials is None:
+            raise CallError(
+                'Only users with full administrative privileges can access internally ran jobs', errno.EPERM,
+            )
+
         if job.credentials.user['username'] == credential.user['username']:
             return job
 

tests/api2/test_job_logs.py

@@ -1,27 +1,58 @@
+import errno
+
+import pytest
 import requests
 
+from middlewared.service_exception import CallError
 from middlewared.test.integration.assets.account import unprivileged_user_client
-from middlewared.test.integration.utils import mock, url
+from middlewared.test.integration.utils import call, mock, url
+
+
+@pytest.fixture(scope="module")
+def c():
+    with unprivileged_user_client(allowlist=[{"method": "CALL", "resource": "test.test1"}]) as c:
+        yield c
 
 
-def test_job_download_logs():
+def test_job_download_logs(c):
     with mock("test.test1", """    
         from middlewared.service import job
 
         @job(logs=True)
         def mock(self, job, *args):
             job.logs_fd.write(b'Job logs')
     """):
-        with unprivileged_user_client(allowlist=[{"method": "CALL", "resource": "test.test1"}]) as c:
-            jid = c.call("test.test1")
+        jid = c.call("test.test1")
+
+        c.call("core.job_wait", jid, job=True)
+
+        path = c.call("core.job_download_logs", jid, 'logs.txt')
 
-            c.call("core.job_wait", jid, job=True)
+        r = requests.get(f"{url()}{path}")
+        r.raise_for_status()
+
+        assert r.headers["Content-Disposition"] == "attachment; filename=\"logs.txt\""
+        assert r.headers["Content-Type"] == "application/octet-stream"
+        assert r.text == "Job logs"
+
+
+def test_job_download_logs_unprivileged_downloads_internal_logs(c):
+    with mock("test.test1", """
+        def mock(self, *args):
+            job = self.middleware.call_sync("test.test2")
+            job.wait_sync(raise_error=True)
+            return job.id
+    """):
+        with mock("test.test2", """
+            from middlewared.service import job
 
-            path = c.call("core.job_download_logs", jid, 'logs.txt')
+            @job(logs=True)
+            def mock(self, job, *args):
+                job.logs_fd.write(b'Job logs')
+        """):
+            jid = call("test.test1")
 
-            r = requests.get(f"{url()}{path}")
-            r.raise_for_status()
+            with pytest.raises(CallError) as ve:
+                c.call("core.job_download_logs", jid, 'logs.txt')
 
-            assert r.headers["Content-Disposition"] == "attachment; filename=\"logs.txt\""
-            assert r.headers["Content-Type"] == "application/octet-stream"
-            assert r.text == "Job logs"
+            assert ve.value.errno == errno.EPERM