Attempt to use null separation when invoking 'env'

This approach eliminates the heuristics for handling newlines in
environment variables on platforms where 'env' supports null separation
via the '-0' flag. The 'sh' shell will first attempt to use null
separation and will fall back to the prior behavior.

colcon_core/shell/__init__.py

@@ -332,6 +332,40 @@ async def get_command_environment(task_name, build_base, dependencies):
         'Could not find a shell extension for the command environment')
 
 
+async def get_null_separated_environment_variables(
+    cmd, *, cwd=None, shell=True,
+):
+    """
+    Get null-separated environment variables from the output of the command.
+
+    :param args: the sequence of program arguments
+    :param cwd: the working directory for the subprocess
+    :param shell: whether to use the shell as the program to execute
+    :rtype: dict
+    """
+    encoding = locale.getpreferredencoding()
+    output = await check_output(cmd, cwd=cwd, shell=shell)
+    env = OrderedDict()
+    for kvp in output.split(b'\0'):
+        kvp = kvp.rstrip()
+        if not kvp:
+            continue
+        try:
+            parts = kvp.decode(encoding).split('=', 1)
+        except UnicodeDecodeError:
+            kvp_replaced = kvp.decode(encoding=encoding, errors='replace')
+            logger.warning(
+                'Failed to decode line from the environment using the '
+                f"encoding '{encoding}': {kvp_replaced}")
+            continue
+        if len(parts) != 2:
+            # skip lines which don't contain an equal sign
+            continue
+        env[parts[0]] = parts[1]
+    assert len(env) > 0, "The environment shouldn't be empty"
+    return env
+
+
 async def get_environment_variables(cmd, *, cwd=None, shell=True):
     """
     Get the environment variables from the output of the command.

colcon_core/shell/sh.py

@@ -10,6 +10,7 @@
 from colcon_core.prefix_path import get_chained_prefix_path
 from colcon_core.shell import check_dependency_availability
 from colcon_core.shell import get_environment_variables
+from colcon_core.shell import get_null_separated_environment_variables
 from colcon_core.shell import logger
 from colcon_core.shell import ShellExtensionPoint
 from colcon_core.shell.template import expand_template
@@ -150,8 +151,13 @@ async def generate_command_environment(  # noqa: D102
             hook_path,
             {'dependencies': dependencies})
 
-        cmd = ['.', str(hook_path), '&&', 'env']
-        env = await get_environment_variables(cmd, cwd=str(build_base))
+        cmd = ['.', str(hook_path), '&&', 'env', '-0']
+        try:
+            env = await get_null_separated_environment_variables(
+                cmd, cwd=str(build_base))
+        except Exception:  # noqa: B902
+            cmd.pop()
+            env = await get_environment_variables(cmd, cwd=str(build_base))
 
         # write environment variables to file for debugging
         env_path = build_base / ('colcon_command_prefix_%s.sh.env' % task_name)

test/test_shell.py

@@ -19,6 +19,7 @@
 from colcon_core.shell import get_command_environment
 from colcon_core.shell import get_environment_variables
 from colcon_core.shell import get_find_installed_packages_extensions
+from colcon_core.shell import get_null_separated_environment_variables
 from colcon_core.shell import get_shell_extensions
 from colcon_core.shell import ShellExtensionPoint
 from colcon_core.shell.installed_packages import IsolatedInstalledPackageFinder
@@ -156,6 +157,42 @@ async def check_output(cmd, **kwargs):
     assert 'DECODE_ERROR=' in warn.call_args[0][0]
 
 
+def test_get_null_separated_environment_variables():
+    cmd = [
+        sys.executable, '-c',
+        r'print("FOO\0NAME=value\nSOMETHING\0NAME2=value with spaces'
+        r'\0NAME3=NAME4\nNAME5=NAME6")']
+
+    coroutine = get_null_separated_environment_variables(cmd, shell=False)
+    env = run_until_complete(coroutine)
+
+    assert len(env.keys()) == 3
+    assert 'NAME' in env.keys()
+    assert env['NAME'] == f'value{os.linesep}SOMETHING'
+    assert 'NAME2' in env.keys()
+    assert env['NAME2'] == 'value with spaces'
+    assert 'NAME3' in env.keys()
+    assert env['NAME3'] == f'NAME4{os.linesep}NAME5=NAME6'
+
+    # test with environment strings which isn't decodable
+    async def check_output(cmd, **kwargs):
+        return b'DECODE_ERROR=\x81\nNAME=value'
+    with patch('colcon_core.shell.check_output', side_effect=check_output):
+        with patch('colcon_core.shell.logger.warning') as warn:
+            coroutine = get_environment_variables(['not-used'], shell=False)
+            env = run_until_complete(coroutine)
+
+    assert len(env.keys()) == 1
+    assert 'NAME' in env.keys()
+    assert env['NAME'] == 'value'
+    # the raised decode error is catched and results in a warning message
+    assert warn.call_count == 1
+    assert len(warn.call_args[0]) == 1
+    assert warn.call_args[0][0].startswith(
+        "Failed to decode line from the environment using the encoding '")
+    assert 'DECODE_ERROR=' in warn.call_args[0][0]
+
+
 class Extension3(ShellExtensionPoint):
     PRIORITY = 105