marketplace-ssh-2014/new-server.rkt

285 lines
11 KiB
Racket

#lang racket/base
;; (Temporary) example client and server
(require racket/set)
(require racket/match)
(require racket/contract)
(require (only-in racket/port peek-bytes-avail!-evt))
(require "cook-port.rkt")
(require "sandboxes.rkt")
(require "ssh-numbers.rkt")
(require "ssh-transport.rkt")
(require "ssh-session.rkt")
(require "ssh-channel.rkt")
(require "ssh-message-types.rkt")
(require "ssh-exceptions.rkt")
(require "marketplace-support.rkt")
(define (main)
(ground-vm (timer-driver)
(tcp-driver)
(tcp-spy)
(spawn #:debug-name 'ssh-tcp-listener #:child listener)))
(define listener
(transition/no-state
(endpoint #:subscriber (tcp-channel ? (tcp-listener 2322) ?)
#:observer
#:conversation r
#:on-presence (session-vm r))))
;;---------------------------------------------------------------------------
(define (check-remote-identification! peer-identification-string)
(define required-peer-identification-regex #rx"^SSH-2\\.0-.*")
;; Each identification string is both a cleartext indicator that
;; we've reached some notion of the right place and also input to
;; the hash function used during D-H key exchange.
(when (not (regexp-match required-peer-identification-regex
peer-identification-string))
(error 'ssh-session
"Invalid peer identification string ~v"
peer-identification-string)))
(define (spy marker)
(define (dump what message)
(write `(,marker ,what ,message))
(newline)
(flush-output)
(void))
(list
(endpoint #:subscriber (wild) #:everything
#:role r
#:on-presence (dump 'arrived r)
#:on-absence (dump 'departed r)
[message (dump 'message message)])
(endpoint #:publisher (wild) #:everything
#:role r
#:on-presence (dump 'arrived r)
#:on-absence (dump 'departed r)
[message (dump 'message message)])))
(define-syntax-rule (wait-as my-orientation topic action ...)
(endpoint my-orientation topic #:observer
#:let-name endpoint-name
#:state state
#:on-presence (begin (printf "WAIT ENDED: ~v\n" topic)
(sequence-actions (transition state
(delete-endpoint endpoint-name)
action ...)))))
(define (session-vm new-conversation)
(match-define (tcp-channel remote-addr local-addr _) new-conversation)
(define local-identification #"SSH-2.0-RacketSSH_0.0")
(define (issue-identification-string)
(at-meta-level
(send-message (tcp-channel local-addr remote-addr
(bytes-append local-identification #"\r\n")))))
(define (read-handshake-and-become-reader)
(transition 'handshake-is-stateless ;; but, crucially, the ssh-reader proper isn't!
(at-meta-level
(endpoint #:subscriber (tcp-channel remote-addr local-addr ?)
#:name 'socket-reader
#:state state
[(tcp-channel _ _ (? eof-object?))
(transition state (quit))]
[(tcp-channel _ _ (? bytes? remote-identification))
(begin
(check-remote-identification! remote-identification)
(sequence-actions (transition state)
;; First, set the incoming mode to bytes.
(at-meta-level
(send-feedback (tcp-channel remote-addr local-addr (tcp-mode 'bytes))))
;; Then initialise the reader, switching to packet-reading mode.
(lambda (ignored-state) (ssh-reader new-conversation))
;; Finally, spawn the remaining processes and issue the initial credit to the reader.
(spawn #:debug-name 'ssh-writer
#:child (ssh-writer new-conversation)
;; TODO: canary: #:exit-signal? #t
)
;; Wait for the reader and writer get started, then tell
;; the reader we are ready for a single packet and spawn
;; the session manager.
(printf "BOO\n")
(wait-as #:subscriber (inbound-packet (wild) (wild) (wild) (wild))
(printf "YAY\n") (flush-output)
(wait-as #:publisher (outbound-packet (wild))
(printf "ALSO YAY\n") (flush-output)
(send-message (inbound-credit 1))
(spawn #:debug-name 'ssh-session
#:pid session-pid
#:child (ssh-session session-pid
local-identification
remote-identification
repl-boot
'server)
;; TODO: canary: #:exit-signal? #t
)))))]))))
(define (exn->outbound-packet reason)
(outbound-packet (ssh-msg-disconnect (exn:fail:contract:protocol-reason-code reason)
(string->bytes/utf-8 (exn-message reason))
#"")))
(define (disconnect-message-required? reason)
(and (exn:fail:contract:protocol? reason)
(not (exn:fail:contract:protocol-originated-at-peer? reason))))
(define (active-exception-handler reason)
;; This is kind of gross: because the absence handler gets invoked
;; several times in a row because of multiple flows intersecting
;; this role, we have to be careful to make the transmission of
;; the disconnection packet idempotent.
;; TODO: this is likely no longer true now we're using exit-signals %%%
(define interesting? (disconnect-message-required? reason))
(transition inert-exception-handler
(when interesting? (send-message (exn->outbound-packet reason)))
(yield #:state state ;; gross
(transition state (at-meta-level (quit #:reason (and interesting? reason)))))))
(define (inert-exception-handler reason)
inert-exception-handler)
(nested-vm #:debug-name (list 'ssh-session-vm new-conversation)
(event-relay 'ssh-event-relay)
(timer-relay 'ssh-timer-relay)
(spy 'SSH)
(issue-identification-string)
;; Expect identification string, then update (!) our inbound
;; subscription handler to switch to packet mode.
(at-meta-level
(send-feedback (tcp-channel remote-addr local-addr (tcp-mode 'lines)))
(send-feedback (tcp-channel remote-addr local-addr (tcp-credit 1))))
(spawn #:debug-name 'ssh-reader
#:child (read-handshake-and-become-reader)
;; TODO: canary: #:exit-signal? #t
)
;; TODO: canary:
;; (spawn #:child
;; (transition active-exception-handler
;; (role (topic-subscriber (exit-signal (wild) (wild)))
;; #:state current-handler
;; #:reason reason
;; #:on-absence (current-handler reason))))
))
;;---------------------------------------------------------------------------
(define (repl-boot user-name)
(list
(event-relay 'app-event-relay)
(spy 'APP)
(at-meta-level
(endpoint #:subscriber (channel-message (channel-stream-name #t (wild)) (wild))
#:conversation (channel-message (channel-stream-name _ cname) _)
#:on-presence (spawn #:debug-name cname #:child (repl-instance user-name cname))))))
;; (repl-instance InputPort OutputPort InputPort OutputPort)
(struct repl-instance-state (c2s-in ;; used by thread to read input from relay
c2s-out ;; used by relay to feed input from remote to the thread
s2c-in ;; used by relay to feed output from thread to remote
s2c-out ;; used by thread to write output to relay
) #:prefab)
(define (repl-instance user-name cname)
(define inbound-stream (channel-stream-name #t cname))
(define outbound-stream (channel-stream-name #f cname))
(define (ch-do action-ctor stream body)
(at-meta-level (action-ctor (channel-message stream body))))
(define (handle-channel-message state body)
(match body
[(channel-stream-request #"pty-req" _)
(match-define (repl-instance-state old-in _ _ old-out) state)
(define-values (cooked-in cooked-out) (cook-io old-in old-out "> "))
(transition (struct-copy repl-instance-state state
[c2s-in cooked-in]
[s2c-out cooked-out])
(ch-do send-feedback inbound-stream (channel-stream-ok)))]
[(channel-stream-notify #"env" _)
;; Don't care
(transition state)]
[(channel-stream-request #"shell" _)
(match-define (repl-instance-state c2s-in _ s2c-in s2c-out) state)
(define buffer-size 1024)
(define dummy-buffer (make-bytes buffer-size))
(define repl-thread (thread (lambda () (repl-shell user-name c2s-in s2c-out))))
(transition state
(ch-do send-feedback inbound-stream (channel-stream-ok))
(endpoint #:subscriber (cons (thread-dead-evt repl-thread) (wild))
[_ (quit #:reason "REPL thread exited")])
(endpoint #:subscriber (cons (peek-bytes-avail!-evt dummy-buffer 0 #f s2c-in) (wild))
;; We're using peek-bytes-avail!-evt rather than
;; read-bytes-avail!-evt because of potential overwriting
;; of the buffer. The overwriting can happen when there's
;; any latency between handling the event and the next
;; firing of the event, since the peek-bytes-avail!-evt
;; will overwrite its buffer next time it's synced on.
#:state state
[(cons _ (? eof-object?))
(let ()
(match-define (repl-instance-state c2s-in c2s-out s2c-in s2c-out) state)
(close-input-port c2s-in)
(close-output-port c2s-out)
(close-input-port s2c-in)
(close-output-port s2c-out)
(transition state (quit)))]
[(cons _ (? number? count))
(transition state
(ch-do send-message outbound-stream (channel-stream-data
(read-bytes count s2c-in))))]))]
[(or (channel-stream-data #"\4") ;; C-d a.k.a EOT
(channel-stream-eof))
(let ()
(close-output-port (repl-instance-state-c2s-out state))
;; ^ this signals the repl thread to exit.
;; Now, wait for it to do so.
(transition state))]
[(channel-stream-data bs)
(write-bytes bs (repl-instance-state-c2s-out state))
(flush-output (repl-instance-state-c2s-out state))
(transition state
(ch-do send-feedback inbound-stream (channel-stream-credit (bytes-length bs))))]
[m
(write `(channel inbound ,m)) (newline)
(transition state)]))
(match (channel-name-type cname)
[#"session"
(define-values (c2s-in c2s-out) (make-pipe))
(define-values (s2c-in s2c-out) (make-pipe))
(transition (repl-instance-state c2s-in c2s-out s2c-in s2c-out)
(at-meta-level
(endpoint #:subscriber (channel-message inbound-stream (wild))
#:state state
#:on-presence (transition state
(ch-do send-feedback inbound-stream (channel-stream-config
(default-packet-limit)
#""))
(ch-do send-feedback inbound-stream (channel-stream-credit 1024)))
[(channel-message _ body)
(handle-channel-message state body)]))
(at-meta-level
(endpoint #:publisher (channel-message outbound-stream (wild))
[m
(begin
(write `(channel outbound ,cname ,m)) (newline)
(void))])))]
[type
(transition/no-state
(at-meta-level (send-message
(channel-message outbound-stream
(channel-stream-open-failure
SSH_OPEN_UNKNOWN_CHANNEL_TYPE
(bytes-append #"Unknown channel type " type))))))]))
;;---------------------------------------------------------------------------
;; TODO: module+
(main)