Add support for environment variable expansion in SSH config

This commit adds support for environment variable substitutions for
select string-valued SSH config options. Any option which previously
supported percent expansion can now also support environment variable
expansion using "${varname}".

Thanks go to Aleksandr Ilin for pointing out that OpenSSH supports
more than just a boolean value for the ForwardAgent config option and
that a string value here supports percent and environment variable
expansion.

Author: ronf

asyncssh/config.py

@@ -41,6 +41,10 @@
 ConfigPaths = Union[None, FilePath, Sequence[FilePath]]
 
 
+_token_pattern = re.compile(r'%(.)')
+_env_pattern = re.compile(r'\${(.*)}')
+
+
 def _exec(cmd: str) -> bool:
     """Execute a command and return if exit status is 0"""
 
@@ -93,36 +97,38 @@ def _set_tokens(self) -> None:
 
         raise NotImplementedError
 
-    def _expand_val(self, value: str) -> str:
-        """Perform percent token expansion on a string"""
+    def _expand_token(self, match):
+        """Expand a percent token reference"""
+
+        try:
+            token = match.group(1)
+            return self._tokens[token]
+        except KeyError:
+            if token == 'd':
+                raise ConfigParseError('Home directory is '
+                                       'not available') from None
+            elif token == 'i':
+                raise ConfigParseError('User id not available') from None
+            else:
+                raise ConfigParseError('Invalid token expansion: ' +
+                                       token) from None
 
-        last_idx = 0
-        result: List[str] = []
+    @staticmethod
+    def _expand_env(match):
+        """Expand an environment variable reference"""
 
-        for match in re.finditer(r'%', value):
-            idx = match.start()
+        try:
+            var = match.group(1)
+            return os.environ[var]
+        except KeyError:
+            raise ConfigParseError('Invalid environment expansion: ' +
+                                   var) from None
 
-            if idx < last_idx:
-                continue
+    def _expand_val(self, value: str) -> str:
+        """Perform percent token and environment expansion on a string"""
 
-            try:
-                token = value[idx+1]
-                result.extend([value[last_idx:idx], self._tokens[token]])
-                last_idx = idx + 2
-            except IndexError:
-                raise ConfigParseError('Invalid token substitution') from None
-            except KeyError:
-                if token == 'd':
-                    raise ConfigParseError('Home directory is '
-                                           'not available') from None
-                elif token == 'i':
-                    raise ConfigParseError('User id not available') from None
-                else:
-                    raise ConfigParseError('Invalid token substitution: ' +
-                                           value[idx+1]) from None
-
-        result.append(value[last_idx:])
-        return ''.join(result)
+        return _env_pattern.sub(self._expand_env,
+                                _token_pattern.sub(self._expand_token, value))
 
     def _include(self, option: str, args: List[str]) -> None:
         """Read config from a list of other config files"""

tests/test_config.py

@@ -514,12 +514,25 @@ def test_invalid_percent_expansion(self):
 
         for desc, config_data in (
                 ('Bad token in hostname', 'Hostname %p'),
-                ('Invalid token', 'IdentityFile %x'),
-                ('Percent at end', 'IdentityFile %')):
+                ('Invalid token', 'IdentityFile %x')):
             with self.subTest(desc):
                 with self.assertRaises(asyncssh.ConfigParseError):
                     self._parse_config(config_data)
 
+    def test_env_expansion(self):
+        """Test environment variable expansion"""
+
+        config = self._parse_config('RemoteCommand ${HOME}/.ssh')
+
+        self.assertEqual(config.get('RemoteCommand'), './.ssh')
+
+    def test_invalid_env_expansion(self):
+        """Test invalid environment variable expansion"""
+
+        with self.assertRaises(asyncssh.ConfigParseError):
+            self._parse_config('RemoteCommand ${XXX}')
+
+
 class _TestServerConfig(_TestConfig):
     """Unit tests for server config objects"""