[3006.x] Populate user data / name within runners and for runner events on the event bus

changelog/63148.fixed.md

@@ -0,0 +1 @@
+User responsible for the runner is now correctly reported in the events on the event bus for the runner.

salt/client/mixins.py

@@ -246,6 +246,8 @@ def cmd(
             self.functions[fun], arglist, pub_data
         )
         low = {"fun": fun, "arg": args, "kwarg": kwargs}
+        if "user" in pub_data:
+            low["__user__"] = pub_data["user"]
         return self.low(fun, low, print_event=print_event, full_return=full_return)
 
     @property

salt/modules/saltutil.py

@@ -1730,8 +1730,10 @@ def runner(
         arg = []
     if kwarg is None:
         kwarg = {}
+    pub_data = {}
     jid = kwargs.pop("__orchestration_jid__", jid)
     saltenv = kwargs.pop("__env__", saltenv)
+    pub_data["user"] = kwargs.pop("__pub_user", "UNKNOWN")
     kwargs = salt.utils.args.clean_kwargs(**kwargs)
     if kwargs:
         kwarg.update(kwargs)
@@ -1760,7 +1762,12 @@ def runner(
         )
 
     return rclient.cmd(
-        name, arg=arg, kwarg=kwarg, print_event=False, full_return=full_return
+        name,
+        arg=arg,
+        pub_data=pub_data,
+        kwarg=kwarg,
+        print_event=False,
+        full_return=full_return,
     )
 
 

salt/runners/state.py

@@ -101,6 +101,16 @@ def orchestrate(
        salt-run state.orchestrate webserver pillar_enc=gpg pillar="$(cat somefile.json)"
 
     """
+
+    try:
+        orig_user = __opts__["user"]
+        __opts__["user"] = __user__
+        log.debug(
+            f"changed opts user from original '{orig_user}' to global user '{__user__}'"
+        )
+    except NameError:
+        log.debug("unable to find global user __user__")
+
     if pillar is not None and not isinstance(pillar, dict):
         raise SaltInvocationError("Pillar data must be formatted as a dictionary")
     __opts__["file_client"] = "local"

salt/state.py

@@ -2293,6 +2293,7 @@ def call(self, low, chunks=None, running=None, retries=1):
             initial_ret={"full": state_func_name},
             expected_extra_kws=STATE_INTERNAL_KEYWORDS,
         )
+
         inject_globals = {
             # Pass a copy of the running dictionary, the low state chunks and
             # the current state dictionaries.
@@ -2302,6 +2303,7 @@ def call(self, low, chunks=None, running=None, retries=1):
             "__running__": immutabletypes.freeze(running) if running else {},
             "__instance_id__": self.instance_id,
             "__lowstate__": immutabletypes.freeze(chunks) if chunks else {},
+            "__user__": self.opts.get("user", "UNKNOWN"),
         }
 
         if "__env__" in low:

salt/states/saltmod.py

@@ -125,7 +125,7 @@ def state(
     subset=None,
     orchestration_jid=None,
     failhard=None,
-    **kwargs
+    **kwargs,
 ):
     """
     Invoke a state run on a given target
@@ -454,7 +454,7 @@ def function(
     batch=None,
     subset=None,
     failhard=None,
-    **kwargs
+    **kwargs,
 ):  # pylint: disable=unused-argument
     """
     Execute a single module function on a remote minion via salt or salt-ssh
@@ -780,6 +780,14 @@ def runner(name, **kwargs):
         log.debug("Unable to fire args event due to missing __orchestration_jid__")
         jid = None
 
+    try:
+        kwargs["__pub_user"] = __user__
+        log.debug(
+            f"added __pub_user to kwargs using dunder user '{__user__}', kwargs '{kwargs}'"
+        )
+    except NameError:
+        log.warning("unable to find user for fire args event due to missing __user__")
+
     if __opts__.get("test", False):
         ret = {
             "name": name,
@@ -899,7 +907,7 @@ def call_runner(runner_config):
             __orchestration_jid__=jid,
             __env__=__env__,
             full_return=True,
-            **(runner_config.get("kwarg", {}))
+            **(runner_config.get("kwarg", {})),
         )
 
     try:

tests/pytests/integration/runners/state/orchestrate/test_events.py

@@ -4,16 +4,93 @@
 import concurrent.futures
 import functools
 import json
+import logging
 import time
 
+import attr
 import pytest
+from saltfactories.utils import random_string
 
 import salt.utils.jid
 import salt.utils.platform
+import salt.utils.pycrypto
 
-pytestmark = [
-    pytest.mark.slow_test,
-]
+log = logging.getLogger(__name__)
+
+
+@attr.s(kw_only=True, slots=True)
+class TestMasterAccount:
+    username = attr.ib()
+    password = attr.ib()
+    _delete_account = attr.ib(init=False, repr=False, default=False)
+
+    @username.default
+    def _default_username(self):
+        return random_string("account-", uppercase=False)
+
+    @password.default
+    def _default_password(self):
+        return random_string("pwd-", size=8)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        pass
+
+
+@pytest.fixture(scope="session")
+def salt_auth_account_m_factory():
+    return TestMasterAccount(username="saltdev-auth-m")
+
+
+@pytest.fixture(scope="module")
+def salt_auth_account_m(salt_auth_account_m_factory):
+    with salt_auth_account_m_factory as account:
+        yield account
+
+
+@pytest.fixture(scope="module")
+def runner_master_config(salt_auth_account_m):
+    return {
+        "external_auth": {
+            "pam": {salt_auth_account_m.username: [{"*": [".*"]}, "@runner", "@wheel"]}
+        }
+    }
+
+
+@pytest.fixture(scope="module")
+def runner_salt_master(salt_factories, runner_master_config):
+    factory = salt_factories.salt_master_daemon(
+        "runner-master", defaults=runner_master_config
+    )
+    with factory.started():
+        yield factory
+
+
+@pytest.fixture(scope="module")
+def runner_salt_run_cli(runner_salt_master):
+    return runner_salt_master.salt_run_cli()
+
+
+@pytest.fixture(scope="module")
+def runner_salt_call_cli(runner_salt_minion):
+    return runner_salt_minion.salt_call_cli()
+
+
+@pytest.fixture(scope="module")
+def runner_add_user(runner_salt_run_cli, salt_auth_account_m):
+    ## create user on master to use
+    ret = runner_salt_run_cli.run("salt.cmd", "user.add", salt_auth_account_m.username)
+    assert ret.returncode == 0
+
+    yield
+
+    ## remove user on master
+    ret = runner_salt_run_cli.run(
+        "salt.cmd", "user.delete", salt_auth_account_m.username
+    )
+    assert ret.returncode == 0
 
 
 def test_state_event(salt_run_cli, salt_cli, salt_minion):
@@ -335,3 +412,106 @@ def test_orchestration_onchanges_and_prereq(
                 # After the file was created, running again in test mode should have
                 # shown no changes.
                 assert not state_data["changes"]
+
+
+@pytest.mark.slow_test
+@pytest.mark.skip_if_not_root
+@pytest.mark.skip_on_windows
+@pytest.mark.skip_on_darwin
+def test_unknown_in_runner_event(
+    runner_salt_run_cli,
+    runner_salt_master,
+    salt_minion,
+    salt_auth_account_m,
+    runner_add_user,
+    event_listener,
+):
+    """
+    Test to confirm that the ret event for the orchestration contains the
+    jid for the jobs spawned.
+    """
+    file_roots_base_dir = runner_salt_master.config["file_roots"]["base"][0]
+    test_top_file_contents = """
+    base:
+      '{minion_id}':
+        - {file_roots}
+    """.format(
+        minion_id=salt_minion.id, file_roots=file_roots_base_dir
+    )
+    test_init_state_contents = """
+    always-passes-with-any-kwarg:
+      test.nop:
+        - name: foo
+        - something: else
+        - foo: bar
+    always-passes:
+      test.succeed_without_changes:
+        - name: foo
+    always-changes-and-succeeds:
+      test.succeed_with_changes:
+        - name: foo
+    {{slspath}}:
+      test.nop
+    """
+    test_orch_contents = """
+    test_highstate:
+      salt.state:
+        - tgt: {minion_id}
+        - highstate: True
+    test_runner_metasyntetic:
+      salt.runner:
+        - name: test.metasyntactic
+        - locality: us
+    """.format(
+        minion_id=salt_minion.id
+    )
+    with runner_salt_master.state_tree.base.temp_file(
+        "top.sls", test_top_file_contents
+    ), runner_salt_master.state_tree.base.temp_file(
+        "init.sls", test_init_state_contents
+    ), runner_salt_master.state_tree.base.temp_file(
+        "orch.sls", test_orch_contents
+    ):
+        ret = runner_salt_run_cli.run(
+            "salt.cmd", "shadow.gen_password", salt_auth_account_m.password
+        )
+        assert ret.returncode == 0
+
+        gen_pwd = ret.stdout
+        ret = runner_salt_run_cli.run(
+            "salt.cmd", "shadow.set_password", salt_auth_account_m.username, gen_pwd
+        )
+        assert ret.returncode == 0
+
+        jid = salt.utils.jid.gen_jid(runner_salt_master.config)
+        start_time = time.time()
+
+        ret = runner_salt_run_cli.run(
+            "--jid",
+            jid,
+            "-a",
+            "pam",
+            "--username",
+            salt_auth_account_m.username,
+            "--password",
+            salt_auth_account_m.password,
+            "state.orchestrate",
+            "orch",
+        )
+        assert not ret.stdout.startswith("Authentication failure")
+
+        expected_new_event_tag = "salt/run/*/new"
+        event_pattern = (runner_salt_master.id, expected_new_event_tag)
+        found_events = event_listener.get_events([event_pattern], after_time=start_time)
+
+        for event in found_events:
+            if event.data["fun"] == "runner.test.metasyntactic":
+                assert event.data["user"] == salt_auth_account_m.username
+
+        expected_ret_event_tag = "salt/run/*/ret"
+        event_pattern = (runner_salt_master.id, expected_ret_event_tag)
+        found_events = event_listener.get_events([event_pattern], after_time=start_time)
+
+        for event in found_events:
+            if event.data["fun"] == "runner.test.metasyntactic":
+                assert event.data["user"] == salt_auth_account_m.username