Return type of create_process is partially unknown

[MBradbury]

When type checking a program using create_process a warning will be shown that its return type is partially unknown.

This is because create_process (https://github.com/ronf/asyncssh/blob/develop/asyncssh/connection.py#L4057) has been annotated with the return type of SSHClientProcess. This error arises as SSHClientProcess is generic and create_process does not specify the string type.

There are likely three options to fix this:

1. Set the return type to be AnyStr, which means that SSHClientProcess will have the same string type as the input parameter.

   -> SSHClientProcess[AnyStr]

2. Set the return type to be str or bytes, but do not tie the type to the same type as input.

   -> SSHClientProcess[Union[str, bytes]]

3. Use overloads to specify the different type depending on the encoding type. This is used by subprocess, so is probably the way to go.

       @overload
       @async_context_manager
       async def create_process(self, *args: object,
                                input: Optional[str] = None,
                                stdin: ProcessSource = PIPE,
                                stdout: ProcessTarget = PIPE,
                                stderr: ProcessTarget = PIPE,
                                bufsize: int = io.DEFAULT_BUFFER_SIZE,
                                send_eof: bool = True, recv_eof: bool = True,
                                encoding: str,
                                **kwargs: object) -> SSHClientProcess[str]: ...
   
       @overload
       @async_context_manager
       async def create_process(self, *args: object,
                                input: Optional[bytes] = None,
                                stdin: ProcessSource = PIPE,
                                stdout: ProcessTarget = PIPE,
                                stderr: ProcessTarget = PIPE,
                                bufsize: int = io.DEFAULT_BUFFER_SIZE,
                                send_eof: bool = True, recv_eof: bool = True,
                                encoding: None = None,
                                **kwargs: object) -> SSHClientProcess[bytes]: ...
   
       # pylint: disable=redefined-builtin
       @async_context_manager
       async def create_process(self, *args,
                                input = None,
                                stdin = PIPE,
                                stdout = PIPE,
                                stderr = PIPE,
                                bufsize = io.DEFAULT_BUFFER_SIZE,
                                send_eof = True, recv_eof = True,
                                **kwargs):

Version information, but this is also currently an issue in develop.

$ python
Python 3.10.12 (main, Jun 11 2023, 05:26:28) [GCC 11.4.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import asyncssh
>>> asyncssh.__version__
'2.14.0'

Using VSCode and PyLance for type checking.

[ronf]

Thanks for the report! I'm not seeing an issue with this when type checking with 'mypy', but I agree the return type should be better qualified as you suggest in your option #1.

I've made this change in commit f67234f in the "develop" branch -- let me know if you have any problem with it.

[MBradbury]

Hello @ronf, it looks like I was incorrect with option 1. As this ends up typing as SSHClientProcess[Any] when input is not provided, which makes my warnings go away, but then leads to some slightly odd SSHWriter[Any] and SSHReader[Any] types. I think Option 3 is the correct way to go.

When @async_context_manager is commented out, the following types correctly.

1. p = await create_process("./proc") types as SSHClientProcess[bytes]
2. p = await create_process("./proc", encoding="utf-8") types as SSHClientProcess[str]
3. p = await create_process("./proc", encoding=None) types as SSHClientProcess[bytes]
4. p = await create_process("./proc", input=None) types as SSHClientProcess[bytes]
5. p = await create_process("./proc", input="hello") is an error as the type is unknown (need to set encoding)
6. p = await create_process("./proc", input="hello", encoding="utf-8") types as SSHClientProcess[str]
7. p = await create_process("./proc", input=b"hello") types as SSHClientProcess[bytes]
8. p = await create_process("./proc", input=b"hello", encoding="utf-8") is an error

The problem is this does not work when the context manager is used, as it defaults to the first overload. This is likely to be due to microsoft/pyright#2521.

    @overload
    @async_context_manager
    async def create_process(self, *args: object,
                             input: None = None,
                             stdin: ProcessSource = PIPE,
                             stdout: ProcessTarget = PIPE,
                             stderr: ProcessTarget = PIPE,
                             bufsize: int = io.DEFAULT_BUFFER_SIZE,
                             send_eof: bool = True, recv_eof: bool = True,
                             encoding: str,
                             **kwargs: object) -> SSHClientProcess[str]: ...

    @overload
    @async_context_manager
    async def create_process(self, *args: object,
                             input: str,
                             stdin: ProcessSource = PIPE,
                             stdout: ProcessTarget = PIPE,
                             stderr: ProcessTarget = PIPE,
                             bufsize: int = io.DEFAULT_BUFFER_SIZE,
                             send_eof: bool = True, recv_eof: bool = True,
                             encoding: str,
                             **kwargs: object) -> SSHClientProcess[str]: ...

    @overload
    @async_context_manager
    async def create_process(self, *args: object,
                             input: None = None,
                             stdin: ProcessSource = PIPE,
                             stdout: ProcessTarget = PIPE,
                             stderr: ProcessTarget = PIPE,
                             bufsize: int = io.DEFAULT_BUFFER_SIZE,
                             send_eof: bool = True, recv_eof: bool = True,
                             encoding: None = None,
                             **kwargs: object) -> SSHClientProcess[bytes]: ...

    @overload
    @async_context_manager
    async def create_process(self, *args: object,
                             input: bytes,
                             stdin: ProcessSource = PIPE,
                             stdout: ProcessTarget = PIPE,
                             stderr: ProcessTarget = PIPE,
                             bufsize: int = io.DEFAULT_BUFFER_SIZE,
                             send_eof: bool = True, recv_eof: bool = True,
                             encoding: None = None,
                             **kwargs: object) -> SSHClientProcess[bytes]: ...

    # pylint: disable=redefined-builtin
    @async_context_manager
    async def create_process(self, *args: object,
                             input: Optional[AnyStr] = None,
                             stdin: ProcessSource = PIPE,
                             stdout: ProcessTarget = PIPE,
                             stderr: ProcessTarget = PIPE,
                             bufsize: int = io.DEFAULT_BUFFER_SIZE,
                             send_eof: bool = True, recv_eof: bool = True,
                             encoding: Optional[str] = None,
                             **kwargs: object) -> SSHClientProcess[AnyStr]:

[ronf]

Ah, yeah - I have never been able to get full typing to work through @async_context_manager. You can see python/typing#898 for some earlier attempts at this. I'll take a look again at your version of this change with ParamSpec and see if things have improved since I last attempted this, though.

Pyright in general is much stricter than mypy and while I've tried to satisfy it where possible, there are a number of "errors" that pyright reports on AsyncSSH that I haven't been able to resolve without much the code uglier and harder to maintain, and pyright doesn't provide the knobs to disable some of these checks (or at least it didn't the last time I attempted this). So, for now, only mypy is fully supported.

[MBradbury]

I see, it looks like I came to the same conclusion as you. I have installed mypy and it looks like it picks up the type as SSHClientProcess[AnyStr].

[ronf]

Thanks - could you include a code snippet which shows the incomplete type in pyright, so I can try experimenting further with it?

[MBradbury]

With my PR, I no longer get an incomplete type with PyLance (which uses pyright).

I have been doing something similar to:

import asyncssh
async with asyncssh.connect(
                    "192.168.0.1",
                    username="user",
                    password="pass",
                    # Disable host key checking
                    known_hosts=None
                ) as conn:
    async with conn.create_process("./proc", encoding="utf-8") as proc:
        proc.stdin.write("input-to-proc\n")
        await proc.stdin.drain()

        out = (await proc.stdout.readline()).strip()

[MBradbury]

For PyLance

With asyncssh-2.14.0 from PyPI:

With the current develop:

With my PR:

For mypy, it doesn't seem to provide a type for the pwn process:

With my PR:

With either asyncssh-2.14.0 from PyPI or develop: no type pop-up is shown for pwn.stdin.write.

[ronf]

Can you try the following patch? It adds in the ParamSpec that I mentioned in a link above (pretty much matching your change in misc.py), but it doesn't do the overload on create_process.

There are a couple of other fixes in here as well that mypy caught after adding the ParamSpec, suggesting it might be working.

--- a/asyncssh/connection.py
+++ b/asyncssh/connection.py
@@ -4796,8 +4796,8 @@ class SSHClientConnection(SSHConnection):

         """

-        return await listen_reverse(host, port, tunnel=self,
-                                    **kwargs) # type: ignore
+        return await listen_reverse(host, port,
+                                    tunnel=self, **kwargs) # type: ignore

     @async_context_manager
     async def forward_local_port_to_path(
@@ -5038,10 +5038,10 @@ class SSHClientConnection(SSHConnection):
         """

         def session_factory(_orig_host: str,
-                            _orig_port: int) -> Awaitable[SSHUNIXSession]:
+                            _orig_port: int) -> Awaitable[SSHTCPSession]:
             """Return an SSHTCPSession used to do remote port forwarding"""

-            return cast(Awaitable[SSHUNIXSession],
+            return cast(Awaitable[SSHTCPSession],
                         self.forward_unix_connection(dest_path))

         self.logger.info('Creating remote TCP forwarder from %s to %s',
@@ -5079,10 +5079,10 @@ class SSHClientConnection(SSHConnection):

         """

-        def session_factory() -> Awaitable[SSHTCPSession[bytes]]:
+        def session_factory() -> Awaitable[SSHUNIXSession[bytes]]:
             """Return an SSHUNIXSession used to do remote path forwarding"""

-            return cast(Awaitable[SSHTCPSession[bytes]],
+            return cast(Awaitable[SSHUNIXSession[bytes]],
                         self.forward_connection(dest_host, dest_port))

         self.logger.info('Creating remote UNIX forwarder from %s to %s',
diff --git a/asyncssh/misc.py b/asyncssh/misc.py
index 22c75d9..f01b9a4 100644
--- a/asyncssh/misc.py
+++ b/asyncssh/misc.py
@@ -29,8 +29,8 @@ from pathlib import Path, PurePath
 from random import SystemRandom
 from types import TracebackType
 from typing import Any, AsyncContextManager, Awaitable, Callable, Dict
-from typing import Generator, Generic, IO, Mapping, Optional, Sequence
-from typing import Tuple, Type, TypeVar, Union, cast, overload
+from typing import Generator, Generic, IO, Mapping, Optional, ParamSpec
+from typing import Sequence, Tuple, Type, TypeVar, Union, cast, overload
 from typing_extensions import Literal, Protocol

 from .constants import DEFAULT_LANG
@@ -288,10 +288,12 @@ class _ACMWrapper(Generic[_ACM]):
         return exit_result


-_ACMCoro = Callable[..., Awaitable[_ACM]]
-_ACMWrapperFunc = Callable[..., _ACMWrapper[_ACM]]
+_ACMParams = ParamSpec('_ACMParams')
+_ACMCoro = Callable[_ACMParams, Awaitable[_ACM]]
+_ACMWrapperFunc = Callable[_ACMParams, _ACMWrapper[_ACM]]

-def async_context_manager(coro: _ACMCoro[_ACM]) -> _ACMWrapperFunc[_ACM]:
+def async_context_manager(coro: _ACMCoro[_ACMParams, _ACM]) -> \
+        _ACMWrapperFunc[_ACMParams, _ACM]:
     """Decorator for functions returning asynchronous context managers

        This decorator can be used on functions which return objects
@@ -305,7 +307,8 @@ def async_context_manager(coro: _ACMCoro[_ACM]) -> _ACMWrapperFunc[_ACM]:
     """

     @functools.wraps(coro)
-    def context_wrapper(*args, **kwargs) -> _ACMWrapper[_ACM]:
+    def context_wrapper(*args: _ACMParams.args,
+                        **kwargs: _ACMParams.kwargs) -> _ACMWrapper[_ACM]:
         """Return an async context manager wrapper for this coroutine"""

         return _ACMWrapper(coro(*args, **kwargs))

[ronf]

Ok - I see now why you were adding the overloads. The above works when "input"is specified, but otherwise the type is not bound, leaving it returning SSHClientProcess[Unknown]. Adding overloads can allow "encoding" to also influence the type, at least when the call passes a string literal or the value None.

I'm also running into a separate issue where I get back the following error when the type ends up as Unknown:

/private/tmp/t.py:9:30 - error: Argument of type "Literal['foo\n']" cannot be assigned to parameter "data" of type "AnyStr@create_process" in function "write"
  Type "Literal['foo\n']" cannot be assigned to type "AnyStr@create_process" (reportGeneralTypeIssues)

This is on pyright 1.1.335 and Python 3.11.6.

It doesn't seem to recognize that a literal string should be allowed when the type is AnyStr. I can work around this by creating a new _AnyStr = Type('_AnyStr', AnyStr), but that's rather ugly.

[ronf]

As I dig into this, I've run into a major problem. AsyncSSH allows the encoding to be selected at the connection level or via config files, and so the default value of this argument is actually the empty tuple, which is a placeholder indicating that the encoding should be pulled from "_options", or set to the default of "utf-8" if not specified there. In this case, though, there's no way to infer whether the SSHClientProcess should be set to use bytes or str, though. It will do the enforcement of this at runtime when data is passed in to write(). So, while cases with encoding set explicitly might benefit from stricter type checking, the common case where encoding is not present in the create_process() call as an explicit argument will have to leave it ambiguous, and also triggers the error I mentioned above.

Assuming the overloads could be make to work, this change seems like it has limited value, pretty much only helping with encoding is set explicitly on the create_process call and input is not set. When encoding is set at the connection level or via Options, this change won't provide any benefit.

[MBradbury]

Thanks for taking look @ronf, it has been greatly appreciated. I did wonder what the default value of tuple was intended to achieve, so that makes sense. Also understandable that the lack of ParamSpec in older Python versions would prevent using it for now. I believe that typing-extensions has a ParamSpec, but I have not tried if it will work effectively with the type checkers.

I’m happy to leave this here. It sounds like with the ability to set encoding via options there will not be a way to type this.

I haven’t seen the error you mentioned above. Can you share your test code?

Potentially one solution could be to type the return with AnyStr. (Option 2 of the original post) but this also feels like a poor option.

[ronf]

I fiddled a bit with the version of ParamSpec in typing_extensions, and it looks like that did get past that hurdle. It still works to import ParamSpec from there even on Python 3.10 and later (at least for now), so it doesn't even require a conditional.

That said, I went back and tried to run mypy on the overloads and I'm getting errors on those. I'll go back and try your version exactly and see if that's still the case, but I'll need to modify it to also handle encryption allowing a value of (), and I'm getting another weird error on that where it seems to get confused between the type of args and the type of encoding in the call to create session:

asyncssh/connection.py:4208: error: Argument 2 to "create_session" of "SSHClientConnection" has incompatible type "*Tuple[object, ...]"; expected "Union[Tuple[], str, None]"  [arg-type]

Here's the test code I've been using:

import asyncio, asyncssh

async def run():
    async with asyncssh.connect('localhost') as conn:
        async with conn.create_process("cat") as proc:
            #reveal_type(proc)
            proc.stdin.write("foo\n")
            print(await proc.stdout.readline(), end='')

asyncio.run(run())

Regarding option 2, I've run into trouble in the past with using Union[str, bytes], which is why I ended up having to create generic versions of several of the classes, so separate versions would exist for type checking depending on whether bytes or str was passed into them. I can experiment with that here, but I would be surprised if that was allowed, especially with other places in the system still using AnyStr.

[ronf]

Just confirmed that I get mypy errors even with your version of the overloads:

asyncssh/connection.py:4138: error: Overloaded function implementation cannot produce return type of signature 1  [misc]
asyncssh/connection.py:4138: error: Overloaded function implementation cannot produce return type of signature 2  [misc]
asyncssh/connection.py:4138: error: Overloaded function implementation cannot produce return type of signature 3  [misc]
asyncssh/connection.py:4138: error: Overloaded function implementation cannot produce return type of signature 4  [misc]

The other error related to calling create_session was there previously but disabled with a # type: ignore. It looks like mypy doesn't handle having a mix of position arguments and a "*args" which inserts others, and the error message references the type of encoding, which is even stranger.

[ronf]

Changing from AnyStr to BytesOrStr didn't help here, and in fact it returned a failure of:

asyncssh/connection.py:4098: error: Invalid type argument value for "SSHClientProcess"  [type-var]

I could switch back to just SSHClientProcess instead of SSHClientProcess[AnyStr] on the create_process return value to avoid that (which is what was in place in 2.14.0), but it seems like that would be moving backward in terms of avoiding unbound types.

For now, I think I've going to abandon this, but I saved a copy of my async_context_manager changes in case there's some reason to try this again in the future. Sorry I don't have better news, but I appreciate the suggestion!

[pyhedgehog]

Maybe you are searching for --strict parameter of mypy?

$ git status
HEAD detached at v2.14.0
nothing to commit, working tree clean
$ mypy asyncssh|wc -l
3
$ mypy --strict asyncssh|wc -l
320

[ronf]

I'm definitely not running with "--strict" here. In fact, to keep the error count manageable, I set "--allow-redefinition". There are a number of places where AsyncSSH allows intentionally flexible typing, but then narrows the type as a function proceeds, and mypy doesn't like redefinitions of that sort.

[ronf]

Closing due to inactivity. Feel free to reopen this or open a new issue if you need anything else.