Validate inventory from command line

src/ansible_runner/__main__.py

@@ -16,6 +16,8 @@
 # specific language governing permissions and limitations
 # under the License.
 #
+from __future__ import annotations
+
 import ast
 import threading
 import traceback
@@ -32,6 +34,7 @@
 import tempfile
 
 from contextlib import contextmanager
+from pathlib import Path
 from uuid import uuid4
 
 import daemon
@@ -45,7 +48,6 @@
 from ansible_runner.utils.capacity import get_cpu_count, get_mem_in_bytes, ensure_uuid
 from ansible_runner.utils.importlib_compat import importlib_metadata
 from ansible_runner.runner import Runner
-from ansible_runner.exceptions import AnsibleRunnerException
 
 VERSION = importlib_metadata.version("ansible_runner")
 
@@ -437,11 +439,8 @@ def role_manager(vargs):
         output.debug(f"using playbook file {playbook}")
 
         if vargs.get('inventory'):
-            inventory_file = os.path.join(vargs.get('private_data_dir'), 'inventory', vargs.get('inventory'))
-            if not os.path.exists(inventory_file):
-                raise AnsibleRunnerException('location specified by --inventory does not exist')
-            kwargs.inventory = inventory_file
-            output.debug(f"using inventory file {inventory_file}")
+            kwargs.inventory = vargs.get('inventory')
+            output.debug(f"using inventory file {kwargs.inventory}")
 
         roles_path = vargs.get('roles_path') or os.path.join(vargs.get('private_data_dir'), 'roles')
         roles_path = os.path.abspath(roles_path)
@@ -519,6 +518,34 @@ def add_args_to_parser(parser, args):
         parser.add_argument(*arg[0], **arg[1])
 
 
+def valid_inventory(private_data_dir: str, inventory: str) -> str | None:
+    """
+    Validate the --inventory value is an actual file or directory.
+
+    The inventory value from the CLI may only be an existing file. Validate it
+    exists. Supplied value may be either be relative to <private_data_dir/inventory/
+    or an absolute path to a file or directory (even outside of private_data_dir).
+    Since ansible itself accepts a file or directory for the inventory, we check
+    for either.
+
+    :return: Absolute path to the valid inventory, or None otherwise.
+    """
+
+    # check if absolute or relative path exists
+    inv = Path(inventory)
+    if inv.exists() and (inv.is_file() or inv.is_dir()):
+        return inv.absolute()
+
+    # check for a file in the pvt_data_dir inventory subdir
+    inv_subdir_path = Path(private_data_dir, 'inventory', inv)
+    if (not inv.is_absolute()
+            and inv_subdir_path.exists()
+            and (inv_subdir_path.is_file() or inv_subdir_path.is_dir())):
+        return inv_subdir_path.absolute()
+
+    return None
+
+
 def main(sys_args=None):
     """Main entry point for ansible-runner executable
 
@@ -789,6 +816,10 @@ def main(sys_args=None):
             parser.exit(status=1, message="The --hosts option can only be used with -m or -r\n")
         if not (vargs.get('module') or vargs.get('role')) and not vargs.get('playbook'):
             parser.exit(status=1, message="The -p option must be specified when not using -m or -r\n")
+        if vargs.get('inventory') and not (abs_inv := valid_inventory(vargs['private_data_dir'], vargs.get('inventory'))):
+            parser.exit(status=1, message="Value for --inventory does not appear to be a valid path.")
+        else:
+            vargs['inventory'] = abs_inv
 
     output.configure()
 

test/integration/test___main__.py

@@ -131,7 +131,8 @@ def test_cmdline_playbook_hosts():
     cmdline('run', 'private_data_dir', '-p', 'fake', '--hosts', 'all')
     with pytest.raises(SystemExit) as exc:
         ansible_runner.__main__.main()
-        assert exc == 1
+
+    assert exc.value.code == 1
 
 
 def test_cmdline_includes_one_option():
@@ -140,7 +141,8 @@ def test_cmdline_includes_one_option():
     cmdline('run', 'private_data_dir')
     with pytest.raises(SystemExit) as exc:
         ansible_runner.__main__.main()
-        assert exc == 1
+
+    assert exc.value.code == 1
 
 
 def test_cmdline_cmdline_override(tmp_path):
@@ -162,3 +164,19 @@ def test_cmdline_cmdline_override(tmp_path):
 
     cmdline('run', str(private_data_dir), '-p', str(playbook), '--cmdline', '-e foo=bar')
     assert ansible_runner.__main__.main() == 0
+
+
+def test_cmdline_invalid_inventory(tmp_path):
+    """
+    Test that an invalid inventory path causes an error.
+    """
+    private_data_dir = tmp_path
+    inv_path = private_data_dir / 'inventory'
+    inv_path.mkdir(parents=True)
+
+    cmdline('run', str(private_data_dir), '-p', 'test.yml', '--inventory', 'badInventoryPath')
+
+    with pytest.raises(SystemExit) as exc:
+        ansible_runner.__main__.main()
+
+    assert exc.value.code == 1

test/integration/test_main.py

@@ -1,14 +1,12 @@
 # -*- coding: utf-8 -*-
 import multiprocessing
 
-from ansible_runner.__main__ import main
+from test.utils.common import iterate_timeout
 
 import pytest
 import yaml
 
-
-from ansible_runner.exceptions import AnsibleRunnerException
-from test.utils.common import iterate_timeout
+from ansible_runner.__main__ import main
 
 
 @pytest.mark.parametrize(
@@ -102,8 +100,8 @@ def test_role_bad_project_dir(tmp_path, project_fixtures):
 @pytest.mark.parametrize('envvars', [
     {'msg': 'hi'},
     {
-        'msg': u'utf-8-䉪ቒ칸ⱷ?噂폄蔆㪗輥',
-        u'蔆㪗輥': u'䉪ቒ칸'
+        'msg': 'utf-8-䉪ቒ칸ⱷ?噂폄蔆㪗輥',
+        '蔆㪗輥': '䉪ቒ칸'
     }],
     ids=['regular-text', 'utf-8-text']
 )
@@ -143,12 +141,12 @@ def test_role_run_inventory(project_fixtures):
 
 
 def test_role_run_inventory_missing(project_fixtures):
-    with pytest.raises(AnsibleRunnerException):
-        main(['run', '-r', 'benthomasson.hello_role',
-              '--hosts', 'testhost',
-              '--roles-path', str(project_fixtures / 'use_role' / 'roles'),
-              '--inventory', 'does_not_exist',
-              str(project_fixtures / 'use_role')])
+    rc = main(['run', '-r', 'benthomasson.hello_role',
+               '--hosts', 'testhost',
+               '--roles-path', str(project_fixtures / 'use_role' / 'roles'),
+               '--inventory', 'does_not_exist',
+               str(project_fixtures / 'use_role')])
+    assert rc == 1
 
 
 def test_role_start(project_fixtures):

test/unit/test__main__.py

@@ -0,0 +1,52 @@
+from ansible_runner.__main__ import valid_inventory
+
+
+def test_valid_inventory_file_in_inventory(tmp_path):
+    """
+    Test a relative file name within inventory subdir.
+    """
+    data_dir = tmp_path / "datadir"
+    inv_dir = data_dir / "inventory"
+    inv_dir.mkdir(parents=True)
+
+    hosts = inv_dir / "hosts.xyz"
+    hosts.touch()
+
+    assert valid_inventory(str(data_dir), "hosts.xyz") == hosts.absolute()
+
+
+def test_valid_inventory_absolute_path_to_file(tmp_path):
+    """
+    Test an absolute path to a file outside of data dir.
+    """
+    data_dir = tmp_path / "datadir"
+    inv_dir = data_dir / "inventory"
+    inv_dir.mkdir(parents=True)
+
+    (tmp_path / "otherdir").mkdir()
+    hosts = tmp_path / "otherdir" / "hosts.xyz"
+    hosts.touch()
+
+    assert valid_inventory(str(data_dir), str(hosts.absolute())) == hosts.absolute()
+
+
+def test_valid_inventory_absolute_path_to_directory(tmp_path):
+    """
+    Test an absolute path to a directory outside of data dir.
+    """
+    data_dir = tmp_path / "datadir"
+    inv_dir = data_dir / "inventory"
+    inv_dir.mkdir(parents=True)
+
+    (tmp_path / "otherdir").mkdir()
+    hosts = tmp_path / "otherdir"
+    hosts.touch()
+
+    assert valid_inventory(str(data_dir), str(hosts.absolute())) == hosts.absolute()
+
+
+def test_valid_inventory_doesnotexist(tmp_path):
+    """
+    Test that a bad inventory path returns False.
+    """
+    assert valid_inventory(str(tmp_path), "doesNotExist") is None