Module: Hop::WssBearer

Defined in:
lib/hop/wss_bearer.rb

Defined Under Namespace

Classes: Admission, BufferedSocket, SocketLease

Constant Summary collapse

GUID = 
"258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
MAX_MESSAGE_BYTES = 
1 << 20
MAX_FRAME_BYTES = 
MAX_MESSAGE_BYTES
MAX_HEADER_BYTES = 
16 << 10
MAX_PENDING_CONNECTIONS = 
64
HANDSHAKE_WORKERS = 
4
HANDSHAKE_TIMEOUT_S = 
5.0
READ_TIMEOUT_S = 
15.0

Class Method Summary collapse

Class Method Details

.accept_key(key) ⇒ Object 



30
 
# File 'lib/hop/wss_bearer.rb', line 30

def self.accept_key(key) = Base64.strict_encode64(Digest::SHA1.digest(key + GUID))

.apply_mask(data, mask) ⇒ Object 



54
55
56
57
58
 
# File 'lib/hop/wss_bearer.rb', line 54

def self.apply_mask(data, mask)
  out = data.dup.b
  out.bytesize.times { |i| out.setbyte(i, out.getbyte(i) ^ mask.getbyte(i % 4)) }
  out
end

.dial(endpoint, wss_url, insecure_tls: false) ⇒ Object 



402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
 
# File 'lib/hop/wss_bearer.rb', line 402

def self.dial(endpoint, wss_url, insecure_tls: false)
  uri = URI.parse(wss_url)
  ctx = OpenSSL::SSL::SSLContext.new
  ctx.verify_mode = insecure_tls ? OpenSSL::SSL::VERIFY_NONE : OpenSSL::SSL::VERIFY_PEER
  raw = TCPSocket.new(uri.host, uri.port || 443, connect_timeout: HANDSHAKE_TIMEOUT_S)
  sock = OpenSSL::SSL::SSLSocket.new(raw, ctx)
  sock.sync_close = true
  sock.hostname = uri.host
  deadline = monotonic + HANDSHAKE_TIMEOUT_S
  loop do
    result = sock.connect_nonblock(exception: false)
    case result
    when :wait_readable then wait_io(sock, true, deadline)
    when :wait_writable then wait_io(sock, false, deadline)
    else break
    end
  end
  key = Base64.strict_encode64(Random.bytes(16))
  path = uri.path.to_s.empty? ? "/_hop" : uri.path
  write_all(sock, "GET #{path} HTTP/1.1\r\nHost: #{uri.host}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: #{key}\r\nSec-WebSocket-Version: 13\r\n\r\n")
  _version, status, _headers, rest = read_http_response(sock)
  raise "WS upgrade failed: #{status}" unless status&.include?("101")

  endpoint.register_closer { sock.close rescue nil } # so endpoint#close ends run_link's read loop
  Thread.new { run_link(endpoint, BufferedSocket.new(sock, rest), :dialer, true) }
  sock
end

.encode_frame(payload, mask) ⇒ Object

Raises:

  • (ArgumentError) 


32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/hop/wss_bearer.rb', line 32

def self.encode_frame(payload, mask)
  n = payload.bytesize
  raise ArgumentError, "WebSocket message exceeds 1 MiB" if n > MAX_MESSAGE_BYTES

  header = (+"\x82").b # FIN + binary opcode
  mb = mask ? 0x80 : 0
  if n < 126
    header << (mb | n).chr
  elsif n < 65_536
    header << (mb | 126).chr << [n].pack("n")
  else
    header << (mb | 127).chr << [n].pack("Q>")
  end
  if mask
    mk = Random.bytes(4)
    header << mk << apply_mask(payload, mk)
  else
    header << payload
  end
  header
end

.handle_conn(endpoint, sock, public_url, ttl_secs, lease = nil) ⇒ Object 



353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
 
# File 'lib/hop/wss_bearer.rb', line 353

def self.handle_conn(endpoint, sock, public_url, ttl_secs, lease = nil)
  _method, path, headers, rest = read_http_head(sock)
  if path == "/.well-known/hop"
    body = Hop::Discovery.well_known_body(endpoint, public_url, ttl_secs)
    write_all(sock, "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: #{body.bytesize}\r\nconnection: close\r\n\r\n#{body}")
    false
  elsif path == "/_hop" && headers["upgrade"]&.downcase == "websocket"
    key = headers["sec-websocket-key"] or raise IOError, "missing Sec-WebSocket-Key"
    write_all(sock, "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: #{accept_key(key)}\r\n\r\n")
    buffered = BufferedSocket.new(sock, rest)
    if lease
      Thread.new do
        begin
          run_link(endpoint, buffered, :acceptor, false)
        ensure
          lease.release
        end
      end
      true
    else
      run_link(endpoint, buffered, :acceptor, false)
      false
    end
  else
    write_all(sock, "HTTP/1.1 404 Not Found\r\nconnection: close\r\n\r\n")
    false
  end
end

.monotonic ⇒ Object 



60
 
# File 'lib/hop/wss_bearer.rb', line 60

def self.monotonic = Process.clock_gettime(Process::CLOCK_MONOTONIC)

.next_link ⇒ Object 



28
 
# File 'lib/hop/wss_bearer.rb', line 28

def self.next_link = @seq_mutex.synchronize { @seq += 1 }

.read_exact(sock, n, deadline: monotonic + READ_TIMEOUT_S) ⇒ Object 



100
101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/hop/wss_bearer.rb', line 100

def self.read_exact(sock, n, deadline: monotonic + READ_TIMEOUT_S)
  return "".b if n.zero?

  data = +"".b
  while data.bytesize < n
    chunk = read_some(sock, n - data.bytesize, deadline)
    raise EOFError, "closed" unless chunk

    data << chunk
  end
  data
end

.read_frame(sock) ⇒ Object 



132
133
134
135
 
# File 'lib/hop/wss_bearer.rb', line 132

def self.read_frame(sock)
  _final, opcode, payload = read_frame_part(sock)
  [opcode, payload]
end

.read_frame_part(sock, remaining = MAX_MESSAGE_BYTES, deadline: monotonic + READ_TIMEOUT_S) ⇒ Object

Raises:

  • (IOError) 


113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/hop/wss_bearer.rb', line 113

def self.read_frame_part(sock, remaining = MAX_MESSAGE_BYTES, deadline: monotonic + READ_TIMEOUT_S)
  b0, b1 = read_exact(sock, 2, deadline: deadline).bytes
  raise IOError, "WebSocket extensions are not supported" unless (b0 & 0x70).zero?

  final = (b0 & 0x80) != 0
  opcode = b0 & 0x0F
  masked = (b1 & 0x80) != 0
  len = b1 & 0x7F
  len = read_exact(sock, 2, deadline: deadline).unpack1("n") if len == 126
  len = read_exact(sock, 8, deadline: deadline).unpack1("Q>") if len == 127
  raise IOError, "WebSocket message exceeds 1 MiB" if len > remaining || len > MAX_MESSAGE_BYTES
  raise IOError, "invalid WebSocket control frame" if opcode >= 0x8 && (!final || len > 125)

  mask = masked ? read_exact(sock, 4, deadline: deadline) : nil
  payload = read_exact(sock, len, deadline: deadline)
  payload = apply_mask(payload, mask) if mask
  [final, opcode, payload]
end

.read_http_head(sock) ⇒ Object

Raises:

  • (IOError) 


258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
 
# File 'lib/hop/wss_bearer.rb', line 258

def self.read_http_head(sock)
  deadline = monotonic + HANDSHAKE_TIMEOUT_S
  data = +"".b
  until data.include?("\r\n\r\n")
    room = MAX_HEADER_BYTES + 1 - data.bytesize
    chunk = read_some(sock, [4096, room].min, deadline)
    raise EOFError, "closed" unless chunk

    data << chunk
    raise IOError, "HTTP headers exceed 16 KiB" if data.bytesize > MAX_HEADER_BYTES
  end
  head, rest = data.split("\r\n\r\n", 2)
  lines = head.split("\r\n")
  request = lines.shift&.split(" ", 3)
  raise IOError, "malformed HTTP request line" unless request&.size == 3

  headers = {}
  lines.each do |line|
    key, value = line.split(":", 2)
    headers[key.strip.downcase] = value.strip if value
  end
  [request[0], request[1], headers, rest.to_s.b]
end

.read_http_response(sock) ⇒ Object 



430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
 
# File 'lib/hop/wss_bearer.rb', line 430

def self.read_http_response(sock)
  deadline = monotonic + HANDSHAKE_TIMEOUT_S
  data = +"".b
  until data.include?("\r\n\r\n")
    room = MAX_HEADER_BYTES + 1 - data.bytesize
    chunk = read_some(sock, [4096, room].min, deadline)
    raise EOFError, "closed" unless chunk

    data << chunk
    raise IOError, "HTTP headers exceed 16 KiB" if data.bytesize > MAX_HEADER_BYTES
  end
  head, rest = data.split("\r\n\r\n", 2)
  lines = head.split("\r\n")
  status = lines.shift.to_s
  [status.split(" ", 2).first, status, lines, rest.to_s.b]
end

.read_message(sock) ⇒ Object

Raises:

  • (IOError) 


137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
 
# File 'lib/hop/wss_bearer.rb', line 137

def self.read_message(sock)
  deadline = monotonic + READ_TIMEOUT_S
  final, opcode, payload = read_frame_part(sock, deadline: deadline)
  return [opcode, payload] if opcode >= 0x8
  raise IOError, "expected a binary WebSocket message" unless opcode == 0x2
  return [opcode, payload] if final

  parts = [payload]
  total = payload.bytesize
  until final
    final, continuation, payload = read_frame_part(sock, MAX_MESSAGE_BYTES - total, deadline: deadline)
    raise IOError, "expected a WebSocket continuation frame" unless continuation.zero?

    total += payload.bytesize
    parts << payload
  end
  [opcode, parts.join]
end

.read_some(sock, n, deadline) ⇒ Object 



71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/hop/wss_bearer.rb', line 71

def self.read_some(sock, n, deadline)
  return sock.read(n) unless sock.respond_to?(:read_nonblock)

  loop do
    result = sock.read_nonblock(n, exception: false)
    case result
    when :wait_readable then wait_io(sock, true, deadline)
    when :wait_writable then wait_io(sock, false, deadline)
    else return result
    end
  end
end

.run_link(endpoint, sock, role, mask) ⇒ Object 



156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
 
# File 'lib/hop/wss_bearer.rb', line 156

def self.run_link(endpoint, sock, role, mask)
  link = next_link
  send_fn = lambda do |buf|
    write_all(sock, encode_frame(buf, mask), READ_TIMEOUT_S)
  rescue StandardError
    nil
  end
  endpoint.register_link(link, role, send_fn)
  loop do
    opcode, payload = read_message(sock)
    break if opcode == 0x8

    endpoint.deliver(link, payload) if opcode == 0x2
  end
rescue EOFError, IOError, OpenSSL::SSL::SSLError, SystemCallError
  nil
ensure
  endpoint.link_down(link)
  begin
    sock.close
  rescue StandardError
    nil
  end
end

.serve(endpoint, port, ssl_context, public_url, ttl_secs = 3600) ⇒ Object 



282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
 
# File 'lib/hop/wss_bearer.rb', line 282

def self.serve(endpoint, port, ssl_context, public_url, ttl_secs = 3600)
  tcp = TCPServer.new(port)
  tcp.listen(MAX_PENDING_CONNECTIONS)
  pending = SizedQueue.new(MAX_PENDING_CONNECTIONS)
  admission = Admission.new(MAX_PENDING_CONNECTIONS)
  closing = false
  close_lock = Mutex.new

  close_server = lambda do
    close_lock.synchronize do
      next if closing

      closing = true
      tcp.close rescue nil
      admission.close_all
      pending.close
    end
  end
  endpoint.register_closer(&close_server)

  HANDSHAKE_WORKERS.times do |i|
    Thread.new do
      Thread.current.name = "hop-wss-handshake-#{i}" if Thread.current.respond_to?(:name=)
      loop do
        break if close_lock.synchronize { closing && pending.empty? }

        lease = pending.pop
        break unless lease

        transferred = false
        begin
          sock = ssl_accept(lease.sock, ssl_context)
          lease.replace(sock)
          transferred = handle_conn(endpoint, sock, public_url, ttl_secs, lease)
        rescue StandardError
          nil
        ensure
          unless transferred
            lease.close
            lease.release
          end
        end
      end
    end
  end

  Thread.new do
    loop do
      raw = tcp.accept
      lease = admission.acquire(raw)
      unless lease
        raw.close rescue nil
        next
      end
      begin
        pending.push(lease, true)
      rescue ThreadError
        lease.close
        lease.release
      end
    rescue IOError, Errno::EBADF
      break
    rescue StandardError
      next unless close_lock.synchronize { closing }

      break
    end
  end
  tcp
end

.ssl_accept(raw, ssl_context) ⇒ Object 



241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
 
# File 'lib/hop/wss_bearer.rb', line 241

def self.ssl_accept(raw, ssl_context)
  sock = OpenSSL::SSL::SSLSocket.new(raw, ssl_context)
  sock.sync_close = true
  deadline = monotonic + HANDSHAKE_TIMEOUT_S
  loop do
    result = sock.accept_nonblock(exception: false)
    case result
    when :wait_readable then wait_io(sock, true, deadline)
    when :wait_writable then wait_io(sock, false, deadline)
    else return sock
    end
  end
rescue StandardError
  sock&.close rescue nil
  raise
end

.wait_io(sock, readable, deadline) ⇒ Object

Raises:

  • (IOError) 


62
63
64
65
66
67
68
69
 
# File 'lib/hop/wss_bearer.rb', line 62

def self.wait_io(sock, readable, deadline)
  remaining = deadline - monotonic
  raise IOError, "socket deadline exceeded" unless remaining.positive?

  io = sock.respond_to?(:to_io) ? sock.to_io : sock
  ready = readable ? IO.select([io], nil, nil, remaining) : IO.select(nil, [io], nil, remaining)
  raise IOError, "socket deadline exceeded" unless ready
end

.write_all(sock, data, timeout = HANDSHAKE_TIMEOUT_S) ⇒ Object 



84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/hop/wss_bearer.rb', line 84

def self.write_all(sock, data, timeout = HANDSHAKE_TIMEOUT_S)
  return sock.write(data) unless sock.respond_to?(:write_nonblock)

  deadline = monotonic + timeout
  offset = 0
  while offset < data.bytesize
    result = sock.write_nonblock(data.byteslice(offset..), exception: false)
    case result
    when :wait_readable then wait_io(sock, true, deadline)
    when :wait_writable then wait_io(sock, false, deadline)
    else offset += result
    end
  end
  offset
end