Merge pull request #28 from JuliaWeb/distributednext

Changes for DistributedNext.jl

Author: JamesWrigley

Project.toml

@@ -1,7 +1,7 @@
 name = "LibSSH"
 uuid = "00483490-30f8-4353-8aba-35b82f51f4d0"
 authors = ["James Wrigley <james@puiterwijk.org> and contributors"]
-version = "0.6.1"
+version = "0.7.0"
 
 [deps]
 CEnum = "fa961155-64e5-5f13-b03f-caf6b980ea82"

docs/src/changelog.md

@@ -7,6 +7,24 @@ CurrentModule = LibSSH
 This documents notable changes in LibSSH.jl. The format is based on [Keep a
 Changelog](https://keepachangelog.com).
 
+## [v0.7.0] - 2024-10-25
+
+### Added
+
+- [`Demo.DemoServer`](@ref) now supports passing `allow_auth_none=true` to allow
+  easily setting up passwordless authentication ([#28]).
+
+### Fixed
+
+- Previously the [`Demo.DemoServer`](@ref)'s command execution implementation
+  would only send the command output after it had finished. Now the output gets
+  sent as soon as it's printed by the command ([#28]).
+
+### Changed
+
+- **Breaking**: [`set_channel_callbacks()`](@ref) will remove any existing
+  callbacks ([#28]).
+
 ## [v0.6.1] - 2024-10-20
 
 ### Added

src/callbacks.jl

@@ -248,14 +248,14 @@ $(TYPEDEF)
 Wrapper around `lib.ssh_channel_callbacks_struct`.
 """
 mutable struct ChannelCallbacks
-    cb_struct::Union{lib.ssh_channel_callbacks_struct, Nothing}
     userdata::Any
     functions::Dict{Symbol, Function}
     c_result_types::Dict{Symbol, DataType}
     c_arg_types::Dict{Symbol}
     jl_result_types::Dict{Symbol}
     jl_result_defaults::Dict{Symbol}
     jl_result_to_ctype::Dict{Symbol}
+    cb_struct::lib.ssh_channel_callbacks_struct
 
     @doc """
     $(TYPEDSIGNATURES)
@@ -329,26 +329,30 @@ mutable struct ChannelCallbacks
                               on_write_wontblock::Union{Function, Nothing}=Returns(0),
                               on_open_response::Union{Function, Nothing}=Returns(nothing),
                               on_request_response::Union{Function, Nothing}=Returns(nothing))
-        self = new(nothing, userdata,
+        self = new(userdata,
                    Dict{Symbol, Function}(),
                    Dict{Symbol, DataType}(),
                    Dict{Symbol, Any}(),
                    Dict{Symbol, Any}(),
                    Dict{Symbol, Any}(),
                    Dict{Symbol, Any}())
 
-        self.cb_struct = lib.ssh_channel_callbacks_struct(sizeof(lib.ssh_channel_callbacks_struct), # size (see: ssh_callback_init())
-                                                          pointer_from_objref(self), # userdata points to self
-                                                          C_NULL, C_NULL,
-                                                          C_NULL, C_NULL,
-                                                          C_NULL, C_NULL,
-                                                          C_NULL, C_NULL,
-                                                          C_NULL, C_NULL,
-                                                          C_NULL, C_NULL,
-                                                          C_NULL, C_NULL,
-                                                          C_NULL,
-                                                          C_NULL,
-                                                          C_NULL)
+        cb_struct = lib.ssh_channel_callbacks_struct(sizeof(lib.ssh_channel_callbacks_struct), # size (see: ssh_callback_init())
+                                                     pointer_from_objref(self), # userdata points to self
+                                                     C_NULL, C_NULL,
+                                                     C_NULL, C_NULL,
+                                                     C_NULL, C_NULL,
+                                                     C_NULL, C_NULL,
+                                                     C_NULL, C_NULL,
+                                                     C_NULL, C_NULL,
+                                                     C_NULL, C_NULL,
+                                                     C_NULL,
+                                                     C_NULL,
+                                                     C_NULL)
+        # Call setfield!() explicitly to fully initialize the object so that all
+        # other calls to setproperty!() will work.
+        setfield!(self, :cb_struct, cb_struct)
+
         self.on_data = on_data
         self.on_eof = on_eof
         self.on_close = on_close

src/channel.jl

@@ -284,17 +284,35 @@ end
 """
 $(TYPEDSIGNATURES)
 
-Wrapper around [`lib.ssh_set_channel_callbacks()`](@ref). Will throw a
-[`LibSSHException`](@ref) if setting the callbacks failed.
+Wrapper around [`lib.ssh_set_channel_callbacks()`](@ref) and
+[`lib.ssh_remove_channel_callbacks()`](@ref). Unlike
+[`lib.ssh_set_channel_callbacks()`](@ref) this will replace any existing
+callbacks.
+
+# Throws
+- [`LibSSHException`](@ref): If setting the callbacks failed.
 """
 function set_channel_callbacks(sshchan::SshChannel, callbacks::Callbacks.ChannelCallbacks)
+    if !isnothing(sshchan.callbacks)
+        remove_channel_callbacks(sshchan, sshchan.callbacks)
+    end
+
     ret = lib.ssh_set_channel_callbacks(sshchan, Ref(callbacks.cb_struct))
     if ret != SSH_OK
         throw(LibSSHException("Error when setting channel callbacks: $(ret)"))
     end
     sshchan.callbacks = callbacks
 end
 
+# Undocumented for now because the API for setting callbacks isn't fleshed out yet
+function remove_channel_callbacks(sshchan::SshChannel, callbacks::Callbacks.ChannelCallbacks)
+    ret = lib.ssh_remove_channel_callbacks(sshchan, Ref(callbacks.cb_struct))
+    if ret != SSH_OK
+        throw(LibSSHException("Error when removing channel callbacks: $(ret)"))
+    end
+    sshchan.callbacks = nothing
+end
+
 """
 $(TYPEDSIGNATURES)
 

src/server.jl

@@ -554,7 +554,12 @@ end
 
 function on_auth_none(session, user, client)::ssh.AuthStatus
     _add_log_event!(client, :auth_none, true)
-    return ssh.AuthStatus_Denied
+
+    if client.allow_auth_none
+        client.authenticated = true
+    end
+
+    return client.authenticated ? ssh.AuthStatus_Success : ssh.AuthStatus_Denied
 end
 
 function on_service_request(session, service, client)::Bool
@@ -580,9 +585,9 @@ function on_channel_env_request(session, sshchan, name, value, client)::Bool
 end
 
 function on_channel_exec_request(session, sshchan, command, client)::Bool
-    _add_log_event!(client, :channel_exec_request, command)
+    _add_log_event!(client, :channel_exec_request, "'$command'")
     owning_sshchan = find_unclaimed_channel(client, sshchan)
-    push!(client.channel_operations, CommandExecutor(command, owning_sshchan, client.env))
+    push!(client.channel_operations, CommandExecutor(client, command, owning_sshchan, client.env))
 
     return true
 end
@@ -695,6 +700,7 @@ end
     session::ssh.Session
     verbose::Bool
     password::Union{String, Nothing}
+    allow_auth_none::Bool = false
     authenticated::Bool = false
 
     session_event::Union{ssh.SessionEvent, Nothing} = nothing
@@ -773,6 +779,7 @@ $(TYPEDFIELDS)
     sshchan::Union{ssh.SshChannel, Nothing} = nothing
     verbose::Bool = false
     password::Union{String, Nothing} = nothing
+    allow_auth_none::Bool = false
 
     clients::Vector{Client} = Client[]
 end
@@ -794,13 +801,17 @@ Creates a [`DemoServer`](@ref).
   authentication etc. Useful for high-level debugging. The events can always be
   printed afterwards with [`Demo.print_timeline`](@ref).
 - `password=nothing`: The password to use if password authentication is enabled.
+- `allow_auth_none`: Whether to allow authentication without any credentials
+  being presented.
 - `auth_methods=[AuthMethod_None, AuthMethod_Password]`: A list of
   authentication methods to enable. See [`ssh.AuthMethod`](@ref).
 - `log_verbosity=nothing`: Controls the logging of libssh itself. This could be
   e.g. `lib.SSH_LOG_WARNING` (see the [upstream
   documentation](https://api.libssh.org/stable/group__libssh__log.html#ga06fc87d81c62e9abb8790b6e5713c55b)).
 """
-function DemoServer(port::Int; verbose::Bool=false, password::Union{String, Nothing}=nothing,
+function DemoServer(port::Int; verbose::Bool=false,
+                    password::Union{String, Nothing}=nothing,
+                    allow_auth_none=false,
                     auth_methods=[ssh.AuthMethod_None, ssh.AuthMethod_Password],
                     log_verbosity=ssh.SSH_LOG_NOLOG)
     if ssh.AuthMethod_Password in auth_methods && isnothing(password)
@@ -810,7 +821,7 @@ function DemoServer(port::Int; verbose::Bool=false, password::Union{String, Noth
     key = pki.generate(pki.KeyType_ed25519)
     bind = ssh.Bind(port; auth_methods, key, log_verbosity)
 
-    demo_server = DemoServer(; bind, verbose, password)
+    demo_server = DemoServer(; bind, verbose, password, allow_auth_none)
 
     ssh.set_message_callback(on_message, bind, demo_server)
 
@@ -893,9 +904,10 @@ end
 
 function _handle_client(session::ssh.Session, ds::DemoServer)
     client = Client(; id=length(ds.clients) + 1,
-                     session,
-                     password=ds.password,
-                     verbose=ds.verbose)
+                    session,
+                    password=ds.password,
+                    allow_auth_none=ds.allow_auth_none,
+                    verbose=ds.verbose)
     server_callbacks = ServerCallbacks(client;
                                        on_auth_password=on_auth_password,
                                        on_auth_none=on_auth_none,
@@ -1016,49 +1028,114 @@ end
 
 ## Execute commands
 
+function on_exec_channel_eof(session, sshchan, executor)
+    _add_log_event!(executor.client, :exec_channel_eof, true)
+end
+
+function on_exec_channel_close(session, sshchan, executor)
+    _add_log_event!(executor.client, :exec_channel_close, true)
+end
+
+function on_exec_channel_data(session, sshchan, data, is_stderr, executor)
+    _add_log_event!(executor.client, :exec_channel_data, length(data))
+
+    # Wait for the command to have been started and the pipe to have been opened
+    timedwait(10) do
+        try
+            isopen(executor.stdin)
+        catch
+            false
+        end
+    end
+
+    write(executor.stdin, data)
+    return length(data)
+end
+
 function exec_command(executor)
     sshchan = executor.sshchan
-    cmd_stdout = IOBuffer()
-    cmd_stderr = IOBuffer()
+    cmd_stdout = ChannelBuffer(sshchan, false)
+    cmd_stderr = ChannelBuffer(sshchan, true)
 
     # Start the process and wait for it
     cmd_str = join(Base.shell_split(executor.command), " ")
     cmd = setenv(ignorestatus(`sh -c $(cmd_str)`), executor.env)
-    proc = run(pipeline(cmd; stdout=cmd_stdout, stderr=cmd_stderr); wait=false)
+    proc = run(pipeline(cmd; stdin=executor.stdin, stdout=cmd_stdout, stderr=cmd_stderr); wait=false)
     executor.process = proc
     notify(executor._started_event)
     wait(proc)
 
-    # Write the output to the channel. We first check if the channel is open in
-    # case it's been killed suddenly in the meantime.
-    if isopen(sshchan)
-        write(sshchan, String(take!(cmd_stdout)))
-        write(sshchan, String(take!(cmd_stderr)); stderr=true)
+    close(executor.stdin)
+    close(cmd_stdout)
+    close(cmd_stderr)
 
-        # Clean up
+    # Clean up
+    if isopen(sshchan)
         ssh.channel_request_send_exit_status(sshchan, proc.exitcode)
         closewrite(sshchan)
     end
 
     close(sshchan)
 end
 
+# This is a helper IO type that exists for the sole purpose of asynchronously
+# forwarding output from commands back to the SshChannel. Other containers like
+# IOBuffer aren't thread-safe and can't be used so this implements a minimal,
+# thread-safe IO type based on Channels.
+mutable struct ChannelBuffer <: IO
+    channel::Channel{Vector{UInt8}}
+    sshchan::ssh.SshChannel
+    is_stderr::Bool
+    task::Task
+
+    function ChannelBuffer(sshchan, is_stderr)
+        self = new(Channel{Vector{UInt8}}(), sshchan, is_stderr)
+        self.task = Threads.@spawn for data in self.channel
+            # Write the output to the channel. We first check if the channel is open in
+            # case it's been killed suddenly in the meantime.
+            if isopen(sshchan)
+                write(self.sshchan, data; stderr=self.is_stderr)
+            end
+        end
+        errormonitor(self.task)
+
+        return self
+    end
+end
+
+function Base.write(chbuf::ChannelBuffer, data::Vector{UInt8})
+    put!(chbuf.channel, data)
+    return length(data)
+end
+
+function Base.close(chbuf::ChannelBuffer)
+    close(chbuf.channel)
+    wait(chbuf.task)
+end
+
 @kwdef mutable struct CommandExecutor
+    client::Client
     command::String
     sshchan::ssh.SshChannel
     env::Dict{String, String}
     task::Union{Task, Nothing} = nothing
     process::Union{Base.Process, Nothing} = nothing
+    stdin::Base.PipeEndpoint = Base.PipeEndpoint()
 
     _started_event::Base.Event = Base.Event()
 end
 
-function CommandExecutor(command::String, sshchan::ssh.SshChannel, env)
+function CommandExecutor(client::Client, command::String, sshchan::ssh.SshChannel, env)
     if !sshchan.owning
         throw(ArgumentError("The passed SshChannel is non-owning, CommandExecutor requires an owning SshChannel"))
     end
 
-    executor = CommandExecutor(; command, sshchan, env)
+    executor = CommandExecutor(; client, command, sshchan, env)
+    callbacks = ChannelCallbacks(executor;
+                                 on_data=on_exec_channel_data,
+                                 on_eof=on_exec_channel_eof,
+                                 on_close=on_exec_channel_close)
+    ssh.set_channel_callbacks(sshchan, callbacks)
 
     executor.task = Threads.@spawn try
         exec_command(executor)

test/LibSSHTests.jl

@@ -177,10 +177,12 @@ end
     # https://github.com/JuliaLang/julia/issues/39282
     # Also note that we set `-F none` to disabling reading user config files.
     openssh_cmd = OpenSSH_jll.ssh()
-    ssh_cmd(cmd::Cmd) = ignorestatus(Cmd(`sshpass -p bar $(openssh_cmd.exec) -F none -o NoHostAuthenticationForLocalhost=yes -p 2222 $cmd`; env=openssh_cmd.env))
+    ssh_args = `-F none -o NoHostAuthenticationForLocalhost=yes -p 2222`
+    ssh_cmd(cmd::Cmd) = ignorestatus(Cmd(`sshpass -p bar $(openssh_cmd.exec) $(ssh_args) $cmd`; env=openssh_cmd.env))
+    passwordless_ssh_cmd(cmd::Cmd) = ignorestatus(Cmd(`$(openssh_cmd.exec) $(ssh_args) $cmd`; env=openssh_cmd.env))
 
     @testset "Command execution" begin
-        DemoServer(2222; password="bar") do
+        DemoServer(2222; password="bar", verbose=false) do
             # Test exitcodes
             @test run(ssh_cmd(`foo@localhost exit 0`)).exitcode == 0
             @test run(ssh_cmd(`foo@localhost exit 42`)).exitcode == 42
@@ -192,6 +194,20 @@ end
             cmd_result = run(pipeline(cmd; stdout=cmd_out))
 
             @test strip(String(take!(cmd_out))) == "bar"
+
+            # Test writing data to stdin
+            read_cmd = "read var && echo \$var"
+            cmd_out = IOBuffer()
+            open(ssh_cmd(`foo@localhost $read_cmd`), cmd_out; write=true) do io
+                write(io, "foo\n")
+            end
+            @test String(take!(cmd_out)) == "foo\n"
+        end
+    end
+
+    @testset "allow_auth_none" begin
+        DemoServer(2222; auth_methods=[ssh.AuthMethod_None], allow_auth_none=true) do
+            @test readchomp(passwordless_ssh_cmd(`foo@localhost whoami`)) == username()
         end
     end
 
@@ -281,7 +297,7 @@ end
         @test client.authenticated
 
         # And the command was executed
-        @test client.callback_log[:channel_exec_request] == ["whoami"]
+        @test client.callback_log[:channel_exec_request] == ["'whoami'"]
     end
 
     @testset "Multiple connections" begin

test/runtests.jl

@@ -3,4 +3,4 @@ import LibSSH
 
 include("LibSSHTests.jl")
 
-retest(LibSSH, LibSSHTests)
+retest(LibSSH, LibSSHTests; stats=true)