Only do substitutions for container path conversions with resolved paths (#12313)

* Resolve paths as much as possible before doing replacements

* Move unused method out of main code, test symlink

awx/main/tests/unit/test_tasks.py

@@ -4,6 +4,7 @@ import json
 import os
 import shutil
 import tempfile
+from pathlib import Path
 
 import fcntl
 from unittest import mock
@@ -36,12 +37,23 @@ from awx.main.models.credential import HIDDEN_PASSWORD, ManagedCredentialType
 from awx.main.tasks import jobs, system
 from awx.main.utils import encrypt_field, encrypt_value
 from awx.main.utils.safe_yaml import SafeLoader
-from awx.main.utils.execution_environments import CONTAINER_ROOT, to_host_path
+from awx.main.utils.execution_environments import CONTAINER_ROOT
 
 from awx.main.utils.licensing import Licenser
 from awx.main.constants import JOB_VARIABLE_PREFIXES
 
 
+def to_host_path(path, private_data_dir):
+    """Given a path inside of the EE container, this gives the absolute path
+    on the host machine within the private_data_dir
+    """
+    if not os.path.isabs(private_data_dir):
+        raise RuntimeError('The private_data_dir path must be absolute')
+    if CONTAINER_ROOT != path and Path(CONTAINER_ROOT) not in Path(path).resolve().parents:
+        raise RuntimeError(f'Cannot convert path {path} unless it is a subdir of {CONTAINER_ROOT}')
+    return path.replace(CONTAINER_ROOT, private_data_dir, 1)
+
+
 class TestJobExecution(object):
     EXAMPLE_PRIVATE_KEY = '-----BEGIN PRIVATE KEY-----\nxyz==\n-----END PRIVATE KEY-----'
 

awx/main/tests/unit/utils/test_execution_environments.py

@@ -1,6 +1,10 @@
+import shutil
+import os
+from uuid import uuid4
+
 import pytest
 
-from awx.main.utils.execution_environments import to_container_path, to_host_path
+from awx.main.utils.execution_environments import to_container_path
 
 
 private_data_dir = '/tmp/pdd_iso/awx_xxx'
@@ -10,26 +14,33 @@ private_data_dir = '/tmp/pdd_iso/awx_xxx'
     'container_path,host_path',
     [
         ('/runner', private_data_dir),
-        ('/runner/foo', '{0}/foo'.format(private_data_dir)),
-        ('/runner/foo/bar', '{0}/foo/bar'.format(private_data_dir)),
-        ('/runner{0}'.format(private_data_dir), '{0}{0}'.format(private_data_dir)),
+        ('/runner/foo', f'{private_data_dir}/foo'),
+        ('/runner', f'{private_data_dir}/foobar/..'),  # private_data_dir path needs to be resolved
+        ('/runner/bar', f'{private_data_dir}/bar/foo/..'),
+        ('/runner/foo/bar', f'{private_data_dir}/foo/bar'),
+        (f'/runner{private_data_dir}', f'{private_data_dir}{private_data_dir}'),
     ],
 )
 def test_switch_paths(container_path, host_path):
     assert to_container_path(host_path, private_data_dir) == container_path
-    assert to_host_path(container_path, private_data_dir) == host_path
 
 
-@pytest.mark.parametrize(
-    'container_path',
-    [
-        ('/foobar'),
-        ('/runner/..'),
-    ],
-)
-def test_invalid_container_path(container_path):
-    with pytest.raises(RuntimeError):
-        to_host_path(container_path, private_data_dir)
+def test_symlink_isolation_dir(request):
+    rand_str = str(uuid4())[:8]
+    dst_path = f'/tmp/ee_{rand_str}_symlink_dst'
+    src_path = f'/tmp/ee_{rand_str}_symlink_src'
+
+    def remove_folders():
+        os.unlink(dst_path)
+        shutil.rmtree(src_path)
+
+    request.addfinalizer(remove_folders)
+    os.mkdir(src_path)
+    os.symlink(src_path, dst_path)
+
+    pdd = f'{dst_path}/awx_xxx'
+
+    assert to_container_path(f'{pdd}/env/tmp1234', pdd) == '/runner/env/tmp1234'
 
 
 @pytest.mark.parametrize(

awx/main/utils/execution_environments.py

@@ -58,17 +58,9 @@ def to_container_path(path, private_data_dir):
     """
     if not os.path.isabs(private_data_dir):
         raise RuntimeError('The private_data_dir path must be absolute')
-    if private_data_dir != path and Path(private_data_dir) not in Path(path).resolve().parents:
-        raise RuntimeError(f'Cannot convert path {path} unless it is a subdir of {private_data_dir}')
-    return path.replace(private_data_dir, CONTAINER_ROOT, 1)
-
-
-def to_host_path(path, private_data_dir):
-    """Given a path inside of the EE container, this gives the absolute path
-    on the host machine within the private_data_dir
-    """
-    if not os.path.isabs(private_data_dir):
-        raise RuntimeError('The private_data_dir path must be absolute')
-    if CONTAINER_ROOT != path and Path(CONTAINER_ROOT) not in Path(path).resolve().parents:
-        raise RuntimeError(f'Cannot convert path {path} unless it is a subdir of {CONTAINER_ROOT}')
-    return path.replace(CONTAINER_ROOT, private_data_dir, 1)
+    # due to how tempfile.mkstemp works, we are probably passed a resolved path, but unresolved private_data_dir
+    resolved_path = Path(path).resolve()
+    resolved_pdd = Path(private_data_dir).resolve()
+    if resolved_pdd != resolved_path and resolved_pdd not in resolved_path.parents:
+        raise RuntimeError(f'Cannot convert path {resolved_path} unless it is a subdir of {resolved_pdd}')
+    return str(resolved_path).replace(str(resolved_pdd), CONTAINER_ROOT, 1)