diff --git a/src/middlewared/middlewared/job.py b/src/middlewared/middlewared/job.py
index db9224e1083f..35a09196fc7b 100644
--- a/src/middlewared/middlewared/job.py
+++ b/src/middlewared/middlewared/job.py
@@ -86,6 +86,19 @@ def get(self, item):
     def all(self):
         return self.deque.all()
 
+    def for_username(self, username):
+        out = {}
+        for jid, job in self.all().items():
+            if job.credentials is None or not job.credentials.is_user_session:
+                continue
+
+            if job.credentials.user['username'] != username:
+                continue
+
+            out[jid] = job
+
+        return out
+
     def add(self, job):
         self.handle_lock(job)
         if job.options["lock_queue_size"] is not None:
diff --git a/src/middlewared/middlewared/plugins/account_/privilege_utils.py b/src/middlewared/middlewared/plugins/account_/privilege_utils.py
index 9ee5903a76f9..28b52f990937 100644
--- a/src/middlewared/middlewared/plugins/account_/privilege_utils.py
+++ b/src/middlewared/middlewared/plugins/account_/privilege_utils.py
@@ -1,3 +1,10 @@
+def credential_has_full_admin(credential):
+    if credential.is_user_session and 'FULL_ADMIN' in credential.user['privilege']['roles']:
+       return True
+
+    return credential.allowlist.full_admin
+
+
 def privileges_group_mapping(
     privileges: list,
     group_ids: list,
diff --git a/src/middlewared/middlewared/plugins/auth.py b/src/middlewared/middlewared/plugins/auth.py
index bbc283adce2f..d935d3a64940 100644
--- a/src/middlewared/middlewared/plugins/auth.py
+++ b/src/middlewared/middlewared/plugins/auth.py
@@ -166,9 +166,14 @@ def dump(self):
 
 class TokenSessionManagerCredentials(SessionManagerCredentials):
     def __init__(self, token_manager, token):
+        root_credentials = token.root_credentials()
+
         self.token_manager = token_manager
         self.token = token
-        self.is_user_session = token.root_credentials().is_user_session
+        self.is_user_session = root_credentials.is_user_session
+        if self.is_user_session:
+            self.user = root_credentials.user
+            self.allowlist = root_credentials.allowlist
 
     def is_valid(self):
         return self.token.is_valid()
@@ -183,9 +188,14 @@ def logout(self):
         self.token_manager.destroy(self.token)
 
     def dump(self):
-        return {
+        data = {
             "parent": dump_credentials(self.token.parent_credentials),
         }
+        if self.is_user_session:
+            data["username"] = self.user["username"]
+
+        return data
+
 
 
 def is_internal_session(session):
diff --git a/src/middlewared/middlewared/service/core_service.py b/src/middlewared/middlewared/service/core_service.py
index b70e8feb3a70..bc24e3d93eb6 100644
--- a/src/middlewared/middlewared/service/core_service.py
+++ b/src/middlewared/middlewared/service/core_service.py
@@ -18,6 +18,7 @@
 from middlewared.common.environ import environ_update
 from middlewared.job import Job
 from middlewared.pipe import Pipes
+from middlewared.plugins.account_.privilege_utils import credential_has_full_admin
 from middlewared.schema import accepts, Any, Bool, Datetime, Dict, Int, List, returns, Str
 from middlewared.service_exception import CallError, ValidationErrors
 from middlewared.settings import conf
@@ -28,7 +29,7 @@
 from .compound_service import CompoundService
 from .config_service import ConfigService
 from .crud_service import CRUDService
-from .decorators import filterable, filterable_returns, job, no_auth_required, pass_app, private
+from .decorators import filterable, filterable_returns, job, no_auth_required, no_authz_required, pass_app, private
 from .service import Service
 
 
@@ -134,6 +135,27 @@ def get_tasks(self):
                 'frames': frames,
             }
 
+    def __is_limited_to_own_jobs(self, credential):
+        if credential is None or not credential.is_user_session:
+            return False
+
+        return not credential_has_full_admin(credential)
+
+    def __job_by_credential_and_id(self, credential, job_id):
+        if not self.__is_limited_to_own_jobs(credential):
+            return self.middleware.jobs[job_id]
+
+        if not credential.is_user_session or credential_has_full_admin(credential):
+            return self.middleware.jobs[job_id]
+
+        job = self.middleware.jobs[job_id]
+
+        if job.credentials.user['username'] == credential.user['username']:
+            return job
+
+        raise CallError(f'{job_id}: job is not owned by current session.', errno.EPERM)
+
+    @no_authz_required
     @filterable
     @filterable_returns(Dict(
         'job',
@@ -172,19 +194,34 @@ def get_tasks(self):
         ),
         register=True,
     ))
-    def get_jobs(self, filters, options):
-        """Get the long running jobs."""
+    @pass_app(rest=True)
+    def get_jobs(self, app, filters, options):
+        """
+        Get information about long-running jobs.
+        If authenticated session does not have the FULL_ADMIN role, only
+        jobs owned by the current authenticated session will be returned.
+        """
+        if app and self.__is_limited_to_own_jobs(app.authenticated_credentials):
+            username = app.authenticated_credentials.user['username']
+            jobs = list(self.middleware.jobs.for_username(username).values())
+        else:
+            jobs = list(self.middleware.jobs.all().values())
+
         raw_result = options['extra'].get('raw_result', True)
         jobs = filter_list([
-            i.__encode__(raw_result) for i in list(self.middleware.jobs.all().values())
+            i.__encode__(raw_result) for i in jobs
         ], filters, options)
         return jobs
 
+    @no_authz_required
     @accepts(Int('id'))
     @job()
     async def job_wait(self, job, id_):
-        return await job.wrap(self.middleware.jobs[id_])
+        target_job = self.__job_by_credential_and_id(job.credentials, id_)
 
+        return await job.wrap(target_job)
+
+    @private
     @accepts(Int('id'), Dict(
         'job-update',
         Dict('progress', additional_attrs=True),
@@ -218,9 +255,15 @@ def notify_postinit(self):
         # Let's setup periodic tasks now
         self.middleware._setup_periodic_tasks()
 
+    @no_authz_required
     @accepts(Int('id'))
-    def job_abort(self, id_):
-        job = self.middleware.jobs[id_]
+    @pass_app(rest=True)
+    def job_abort(self, app, id_):
+        if app is None:
+            job = self.middleware.jobs[id_]
+        else:
+            job = self.__job_by_credential_and_id(app.authenticated_credentials, id_)
+
         return job.abort()
 
     def _should_list_service(self, name, service, target):
@@ -593,6 +636,7 @@ async def download(self, app, method, args, filename, buffered):
         return job.id, f'/_download/{job.id}?auth_token={token}'
 
     @private
+    @no_authz_required
     @accepts(Dict('core-job', Int('sleep')))
     @job()
     def job_test(self, job, data):
diff --git a/src/middlewared/middlewared/utils/allowlist.py b/src/middlewared/middlewared/utils/allowlist.py
index 4c2d005ade19..3b400ea18eda 100644
--- a/src/middlewared/middlewared/utils/allowlist.py
+++ b/src/middlewared/middlewared/utils/allowlist.py
@@ -1,10 +1,13 @@
 import fnmatch
 import re
 
+ALLOW_LIST_FULL_ADMIN = {'method': '*', 'resource': '*'}
+
 
 class Allowlist:
     def __init__(self, allowlist):
         self.exact = {}
+        self.full_admin = ALLOW_LIST_FULL_ADMIN in allowlist
         self.patterns = {}
         for entry in allowlist:
             method = entry["method"]
diff --git a/tests/api2/test_account_privilege_role.py b/tests/api2/test_account_privilege_role.py
index df65e731a6a7..98f794be478b 100644
--- a/tests/api2/test_account_privilege_role.py
+++ b/tests/api2/test_account_privilege_role.py
@@ -7,6 +7,7 @@
 from middlewared.test.integration.assets.account import unprivileged_user_client
 from middlewared.test.integration.assets.pool import dataset, snapshot
 from middlewared.test.integration.utils import client
+from time import sleep
 
 logger = logging.getLogger(__name__)
 
@@ -61,6 +62,21 @@ def test_full_admin_role():
     with unprivileged_user_client(["FULL_ADMIN"]) as c:
         c.call("system.general.config")
 
+        # User with FULL_ADMIN role should have something in jobs list
+        assert len(c.call("core.get_jobs")) != 0
+
+        # attempt to wait / cancel job should not fail
+        jid = c.call("core.job_test", {"sleep": 1})
+
+        # TODO: job subscription is currently broken for roles
+        # Once this is fixed we can set `job=True` and remove the sleep
+        wait_job_id = c.call("core.job_wait", jid)
+        sleep(2)
+        result = c.call("core.get_jobs", [["id", "=", wait_job_id]], {"get": True})
+        assert result["state"] == "SUCCESS"
+
+        c.call("core.job_abort", jid)
+
 
 @pytest.mark.parametrize("role,method,params", [
     ("DATASET_READ", "pool.dataset.checksum_choices", []),
@@ -81,6 +97,7 @@ def test_read_role_can_call_method(role, method, params):
     ("filesystem.getacl", ["/"]),
     ("filesystem.acltemplate.by_path", [{"path": "/"}]),
     ("pool.dataset.details", []),
+    ("core.get_jobs", []),
 ])
 def test_readonly_can_call_method(method, params):
     with unprivileged_user_client(["READONLY"]) as c:
@@ -116,3 +133,42 @@ def test_limited_user_auth_token_behavior():
         with client(auth=None) as c2:
             assert c2.call("auth.login_with_token", auth_token)
             c2.call("auth.me")
+            c2.call("core.get_jobs")
+
+
+def test_sharing_manager_jobs():
+    with unprivileged_user_client(["SHARING_MANAGER"]) as c:
+        auth_token = c.call("auth.generate_token")
+        jid = c.call("core.job_test", {"sleep": 1})
+
+        with client(auth=None) as c2:
+            #c.call("core.job_wait", jid, job=True)
+            assert c2.call("auth.login_with_token", auth_token)
+            wait_job_id = c2.call("core.job_wait", jid)
+            sleep(2)
+            result = c2.call("core.get_jobs", [["id", "=", wait_job_id]], {"get": True})
+            assert result["state"] == "SUCCESS"
+            c2.call("core.job_abort", wait_job_id)
+
+
+def test_foreign_job_access():
+    with unprivileged_user_client(["READONLY"]) as unprivileged:
+        with client() as c:
+            job = c.call("core.job_test")
+
+            wait_job_id = unprivileged.call("core.job_wait", job)
+            sleep(2)
+            result = unprivileged.call("core.get_jobs", [["id", "=", wait_job_id]], {"get": True})
+            assert result["state"] != "SUCCESS"
+
+            jobs = unprivileged.call("core.get_jobs", [["id", "=", job]])
+            assert jobs == []
+
+    with unprivileged_user_client(["FULL_ADMIN"]) as unprivileged:
+        with client() as c:
+            job = c.call("core.job_test")
+
+            wait_job_id = unprivileged.call("core.job_wait", job)
+            sleep(2)
+            result = unprivileged.call("core.get_jobs", [["id", "=", wait_job_id]], {"get": True})
+            assert result["state"] == "SUCCESS"