yahns Ruby server user/dev discussion
 help / color / Atom feed
d9c878e0c8ebc2d50acff19585fe39dc34d130cd blob 10714 bytes (raw)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
 
# -*- encoding: binary -*-
# Copyright (C) 2015-2016 all contributors <yahns-public@yhbt.net>
# License: GPL-3.0+ (https://www.gnu.org/licenses/gpl-3.0.txt)
# frozen_string_literal: true

require_relative 'wbuf_lite'

# loaded by yahns/proxy_pass, this relies on Yahns::HttpResponse for
# constants.
module Yahns::HttpResponse # :nodoc:

  # switch and yield
  def proxy_unbuffer(wbuf, nxt = :ignore)
    @state = wbuf
    wbuf.req_res = nil if nxt.nil? && wbuf.respond_to?(:req_res=)
    proxy_wait_next(wbuf.busy == :wait_readable ? Yahns::Queue::QEV_RD :
                    Yahns::Queue::QEV_WR)
    nxt
  end

  def wbuf_alloc(req_res)
    if req_res.proxy_pass.proxy_buffering
      Yahns::Wbuf.new(nil, req_res.alive)
    else
      Yahns::WbufLite.new(req_res)
    end
  end

  # write everything in buf to our client socket (or wbuf, if it exists)
  # it may return a newly-created wbuf or nil
  def proxy_write(wbuf, buf, req_res)
    unless wbuf
      # no write buffer, try to write directly to the client socket
      case rv = String === buf ? kgio_trywrite(buf) : kgio_trywritev(buf)
      when nil then return # done writing buf, likely
      when String, Array # partial write, hope the skb grows
        buf = rv
      when :wait_writable, :wait_readable
        wbuf = req_res.resbuf ||= wbuf_alloc(req_res)
        break
      end while true
    end

    wbuf.wbuf_write(self, buf)
    wbuf.busy ? wbuf : nil
  end

  def proxy_err_response(code, req_res, exc)
    logger = self.class.logger # Yahns::HttpContext#logger
    case exc
    when nil
      logger.error('premature upstream EOF')
    when Kcar::ParserError
      logger.error("upstream response error: #{exc.message}")
    when String
      logger.error(exc)
    else
      Yahns::Log.exception(logger, 'upstream error', exc)
    end
    # try to write something, but don't care if we fail
    Integer === code and
      kgio_trywrite("HTTP/1.1 #{code} #{
                     Rack::Utils::HTTP_STATUS_CODES[code]}\r\n\r\n") rescue nil

    shutdown rescue nil
    @input = @input.close if @input

    # this is safe ONLY because we are in an :ignore state after
    # Fdmap#forget when we got hijacked:
    close

    nil # signal close of req_res from yahns_step in yahns/proxy_pass.rb
  ensure
    wbuf = req_res.resbuf
    wbuf.wbuf_abort if wbuf.respond_to?(:wbuf_abort)
  end

  def wait_on_upstream(req_res)
    req_res.resbuf ||= wbuf_alloc(req_res)
    :wait_readable # self remains in :ignore, wait on upstream
  end

  def proxy_res_headers(res, req_res)
    status, headers = res
    code = status.to_i
    msg = Rack::Utils::HTTP_STATUS_CODES[code]
    env = @hs.env
    have_body = !Rack::Utils::STATUS_WITH_NO_ENTITY_BODY.include?(code) &&
                env['REQUEST_METHOD'] != 'HEAD'.freeze
    flags = MSG_DONTWAIT
    alive = @hs.next? && self.class.persistent_connections
    term = false
    response_headers = req_res.proxy_pass.response_headers

    res = "HTTP/1.1 #{msg ? %Q(#{code} #{msg}) : status}\r\n".dup
    headers.each do |key,value| # n.b.: headers is an Array of 2-element Arrays
      case key
      when /\A(?:Connection|Keep-Alive)\z/i
        next # do not let some upstream headers leak through
      when %r{\AContent-Length\z}i
        term = true
        flags |= MSG_MORE if have_body && value.to_i > 0
      when %r{\ATransfer-Encoding\z}i
        term = true if value =~ /\bchunked\b/i
      end

      # response header mapping
      case val = response_headers[key]
      when :ignore
        next
      when String
        value = val
      end

      res << "#{key}: #{value}\r\n"
    end

    # For now, do not add a Date: header, assume upstream already did it
    # but do not care if they did not

    # chunk the response ourselves if the client supports it,
    # but the backend does not terminate properly
    if alive && ! term && have_body
      if env['HTTP_VERSION'] == 'HTTP/1.1'.freeze
        res << "Transfer-Encoding: chunked\r\n".freeze
      else # we can't persist HTTP/1.0 and HTTP/0.9 w/o Content-Length
        alive = false
      end
    end
    res << (alive ? "Connection: keep-alive\r\n\r\n".freeze
                  : "Connection: close\r\n\r\n".freeze)

    # send the headers
    case rv = kgio_syssend(res, flags)
    when nil # all done, likely
      res.clear
      break
    when String # partial write, highly unlikely
      flags = MSG_DONTWAIT
      res = rv # hope the skb grows
    when :wait_writable, :wait_readable # highly unlikely in real apps
      proxy_write(nil, res, req_res)
      break # keep buffering body...
    end while true
    req_res.alive = alive
    have_body
  end

  def proxy_read_body(tip, kcar, req_res)
    chunk = ''.dup if kcar.chunked?
    len = kcar.body_bytes_left
    rbuf = Thread.current[:yahns_rbuf]
    alive = req_res.alive
    wbuf = req_res.resbuf

    case tmp = tip.shift || req_res.read_nonblock(0x2000, rbuf, exception:false)
    when String
      if len
        kcar.body_bytes_left -= tmp.size # progress for body_eof? => true
      elsif chunk
        kcar.filter_body(chunk, rbuf = tmp) # progress for body_eof? => true
        next if chunk.empty? # call req_res.read_nonblock for more
        tmp = chunk_out(chunk)
      elsif alive # HTTP/1.0 upstream, HTTP/1.1 client
        tmp = chunk_out(tmp)
      # else # HTTP/1.0 upstream, HTTP/1.0 client, do nothing
      end
      wbuf = proxy_write(wbuf, tmp, req_res)
      chunk.clear if chunk
      if Yahns::WbufLite === wbuf
        req_res.proxy_trailers = [ rbuf.dup, tip ] if chunk && kcar.body_eof?
        return proxy_unbuffer(wbuf)
      end
    when nil # EOF
      # HTTP/1.1 upstream, unexpected premature EOF:
      msg = "upstream EOF (#{len} bytes left)" if len
      msg = 'upstream EOF (chunk)' if chunk
      return proxy_err_response(nil, req_res, msg) if msg

      # HTTP/1.0 upstream:
      wbuf = proxy_write(wbuf, "0\r\n\r\n".freeze, req_res) if alive
      req_res.shutdown
      return proxy_unbuffer(wbuf, nil) if Yahns::WbufLite === wbuf
      return proxy_busy_mod(wbuf, req_res)
    when :wait_readable
      return wait_on_upstream(req_res)
    end until kcar.body_eof?

    if chunk
      # tip is an empty array and becomes trailer storage
      req_res.proxy_trailers = [ rbuf.dup, tip ]
      return proxy_read_trailers(kcar, req_res)
    end
    proxy_busy_mod(wbuf, req_res)
  end

  def proxy_read_trailers(kcar, req_res)
    chunk, tlr = req_res.proxy_trailers
    rbuf = Thread.current[:yahns_rbuf]
    wbuf = req_res.resbuf

    until kcar.trailers(tlr, chunk)
      case rv = req_res.read_nonblock(0x2000, rbuf, exception: false)
      when String
        chunk << rv
      when :wait_readable
        return wait_on_upstream(req_res)
      when nil # premature EOF
        return proxy_err_response(nil, req_res, 'upstream EOF (trailers)')
      end # no loop here
    end
    wbuf = proxy_write(wbuf, trailer_out(tlr), req_res)
    return proxy_unbuffer(wbuf) if Yahns::WbufLite === wbuf
    proxy_busy_mod(wbuf, req_res)
  end

  # start streaming the response once upstream is done sending headers to us.
  # returns :wait_readable if we need to read more from req_res
  # returns :ignore if we yield control to the client(self)
  # returns nil if completely done
  def proxy_response_start(res, tip, kcar, req_res)
    have_body = proxy_res_headers(res, req_res)
    tip = tip.empty? ? [] : [ tip ]

    if have_body
      req_res.proxy_trailers = nil # define to avoid uninitialized warnings
      return proxy_read_body(tip, kcar, req_res)
    end

    # unlikely
    wbuf = req_res.resbuf
    return proxy_unbuffer(wbuf) if Yahns::WbufLite === wbuf

    # all done reading response from upstream, req_res will be discarded
    # when we return nil:
    proxy_busy_mod(wbuf, req_res)
  end

  def proxy_response_finish(kcar, req_res)
    req_res.proxy_trailers ? proxy_read_trailers(kcar, req_res)
                           : proxy_read_body([], kcar, req_res)
  end

  def proxy_wait_next(qflags)
    Thread.current[:yahns_fdmap].remember(self)
    # We must allocate a new, empty request object here to avoid a TOCTTOU
    # in the following timeline
    #
    # original thread:                                 | another thread
    # HttpClient#yahns_step                            |
    # r = k.app.call(env = @hs.env)  # socket hijacked into epoll queue
    # <thread is scheduled away>                       | epoll_wait readiness
    #                                                  | ReqRes#yahns_step
    #                                                  | proxy dispatch ...
    #                                                  | proxy_busy_mod
    # ************************** DANGER BELOW ********************************
    #                                                  | HttpClient#yahns_step
    #                                                  | # clears env
    # sees empty env:                                  |
    # return :ignore if env.include?('rack.hijack_io') |
    #
    # In other words, we cannot touch the original env seen by the
    # original thread since it must see the 'rack.hijack_io' value
    # because both are operating in the same Yahns::HttpClient object.
    # This will happen regardless of GVL existence
    hs = Unicorn::HttpRequest.new
    hs.buf.replace(@hs.buf)
    @hs = hs

    # n.b. we may not touch anything in this object once we call queue_mod,
    # another thread is likely to take it!
    Thread.current[:yahns_queue].queue_mod(self, qflags)
  end

  def proxy_busy_mod(wbuf, req_res)
    if wbuf
      # we are completely done reading and buffering the upstream response,
      # but have not completely written the response to the client,
      # yield control to the client socket:
      @state = wbuf
      proxy_wait_next(wbuf.busy == :wait_readable ? Yahns::Queue::QEV_RD :
                      Yahns::Queue::QEV_WR)
      # no touching self after proxy_wait_next, we may be running
      # HttpClient#yahns_step in a different thread at this point
    else
      case http_response_done(req_res.alive)
      when :wait_readable then proxy_wait_next(Yahns::Queue::QEV_RD)
      when :wait_writable then proxy_wait_next(Yahns::Queue::QEV_WR)
      when :close then close
      end
    end
    nil # signal close for ReqRes#yahns_step
  end

  # n.b.: we can use String#size for optimized dispatch under YARV instead
  # of String#bytesize because all the IO read methods return a binary
  # string when given a maximum read length
  def chunk_out(buf)
    [ "#{buf.size.to_s(16)}\r\n", buf, "\r\n".freeze ]
  end

  def trailer_out(tlr)
    "0\r\n#{tlr.map! do |k,v| "#{k}: #{v}\r\n" end.join}\r\n"
  end
end
debug log:

solving d9c878e ...
found d9c878e in https://yhbt.net/yahns.git

yahns Ruby server user/dev discussion

Archives are clonable:
	git clone --mirror https://yhbt.net/yahns-public
	git clone --mirror http://ou63pmih66umazou.onion/yahns-public

Newsgroups are available over NNTP:
	nntp://news.public-inbox.org/inbox.comp.lang.ruby.yahns
	nntp://ou63pmih66umazou.onion/inbox.comp.lang.ruby.yahns

 note: .onion URLs require Tor: https://www.torproject.org/

AGPL code for this site: git clone https://public-inbox.org/ public-inbox