Real-time collaboration for Jupyter Notebooks, Linux Terminals, LaTeX, VS Code, R IDE, and more,
all in one place.

1
# -*- coding: binary -*-
2
require 'thread'
3
require 'rex/socket'
4
module Rex
5
module Services
6

7
###
8
#
9
# This service acts as a local TCP relay whereby clients can connect to a
10
# local listener that forwards to an arbitrary remote endpoint.  Interaction
11
# with the remote endpoint socket requires that it implement the
12
# Rex::IO::Stream interface.
13
#
14
###
15
class LocalRelay
16

17
  include Rex::Service
18

19
  ###
20
  #
21
  # This module is used to extend streams such that they can be associated
22
  # with a relay context and the other side of the stream.
23
  #
24
  ###
25
  module Stream
26

27
    #
28
    # This method is called when the other side has data that has been read
29
    # in.
30
    #
31
    def on_other_data(data)
32
      if relay.on_other_data_proc
33
        relay.on_other_data_proc.call(relay, self, data)
34
      else
35
        put(data)
36
      end
37
    end
38

39
    attr_accessor :relay
40
    attr_accessor :other_stream
41
  end
42

43
  ###
44
  #
45
  # This module is used to extend stream servers such that they can be
46
  # associated with a relay context.
47
  #
48
  ###
49
  module StreamServer
50

51
    #
52
    # This method is called when the stream server receives a local
53
    # connection such that the remote half can be allocated.  The return
54
    # value of the callback should be a Stream instance.
55
    #
56
    def on_local_connection(relay, lfd)
57
      if relay.on_local_connection_proc
58
        relay.on_local_connection_proc.call(relay, lfd)
59
      end
60
    end
61

62
    attr_accessor :relay
63
  end
64

65
  ###
66
  #
67
  # This class acts as an instance of a given local relay.
68
  #
69
  ###
70
  class Relay
71

72
    def initialize(name, listener, opts = {})
73
      self.name                     = name
74
      self.listener                 = listener
75
      self.opts                     = opts
76
      self.on_local_connection_proc = opts['OnLocalConnection']
77
      self.on_conn_close_proc       = opts['OnConnectionClose']
78
      self.on_other_data_proc       = opts['OnOtherData']
79
      if (not $dispatcher['rex'])
80
        register_log_source('rex', $dispatcher['core'], get_log_level('core'))
81
      end
82
    end
83

84
    def shutdown
85
      begin
86
        listener.shutdown if listener
87
      rescue ::Exception
88
      end
89
    end
90

91
    def close
92
      begin
93
        listener.close if listener
94
      rescue ::Exception
95
      end
96
      listener = nil
97
    end
98

99
    attr_reader :name, :listener, :opts
100
    attr_accessor :on_local_connection_proc
101
    attr_accessor :on_conn_close_proc
102
    attr_accessor :on_other_data_proc
103
  protected
104
    attr_writer :name, :listener, :opts
105

106
  end
107

108
  ###
109
  #
110
  # This class acts as an instance of a local relay handling a reverse connection
111
  #
112
  ###
113
  class ReverseRelay < Relay
114

115
    def initialize(name, channel, opts = {})
116

117
      self.name                     = name
118
      self.listener                 = nil
119
      self.opts                     = opts
120
      self.on_local_connection_proc = opts['OnLocalConnection']
121
      self.on_conn_close_proc       = opts['OnConnectionClose']
122
      self.on_other_data_proc       = opts['OnOtherData']
123
      self.channel                  = channel
124

125
      if !$dispatcher['rex']
126
        register_log_source('rex', $dispatcher['core'], get_log_level('core'))
127
      end
128
    end
129

130
    def shutdown
131
      # don't need to do anything here, it's only "close" we care about
132
    end
133

134
    def close
135
      self.channel.close if self.channel
136
      self.channel = nil
137
    end
138

139
    attr_reader :channel
140

141
    protected
142
      attr_writer :channel
143

144
  end
145

146
  #
147
  # Initializes the local tcp relay monitor.
148
  #
149
  def initialize
150
    self.relays       = Hash.new
151
    self.rfds         = Array.new
152
    self.rev_chans    = Array.new
153
    self.relay_thread = nil
154
    self.relay_mutex  = Mutex.new
155
  end
156

157
  ##
158
  #
159
  # Service interface implementors
160
  #
161
  ##
162

163
  #
164
  # Returns the hardcore alias for the local relay service.
165
  #
166
  def self.hardcore_alias(*args)
167
    "__#{args}"
168
  end
169

170
  #
171
  # Returns the alias for this service.
172
  #
173
  def alias
174
    super || "Local Relay"
175
  end
176

177
  #
178
  # Starts the thread that monitors the local relays.
179
  #
180
  def start
181
    if (!self.relay_thread)
182
      self.relay_thread = Rex::ThreadFactory.spawn("LocalRelay", false) {
183
        begin
184
          monitor_relays
185
        rescue ::Exception => e
186
          elog("Error in #{self} monitor_relays", 'rex', error: e)
187
        end
188
      }
189
    end
190
  end
191

192
  #
193
  # Stops the thread that monitors the local relays and destroys all
194
  # listeners, both local and remote.
195
  #
196
  def stop
197
    if (self.relay_thread)
198
      self.relay_thread.kill
199
      self.relay_thread = nil
200
    end
201

202
    self.relay_mutex.synchronize {
203
      self.relays.delete_if { |k, v|
204
        v.shutdown
205
        v.close
206
        true
207
      }
208
    }
209

210
    # make sure we kill off active sockets when we shut down
211
    while self.rfds.length > 0
212
      close_relay_conn(self.rfds.shift) rescue nil
213
    end
214

215
    # we can safely clear the channels array because all of the
216
    # reverse relays were closed down
217
    self.rev_chans.clear
218
    self.relays.clear
219
  end
220

221
  #
222
  # Start a new active listener on the victim ready for reverse connections.
223
  #
224
  def start_reverse_tcp_relay(channel, opts = {})
225
    opts['__RelayType'] = 'tcp'
226
    opts['Reverse'] = true
227

228
    name = "Reverse-#{opts['LocalPort']}"
229

230
    relay = ReverseRelay.new(name, channel, opts)
231

232
    # dirty hack to get "relay" support?
233
    channel.extend(StreamServer)
234
    channel.relay = relay
235

236
    self.relay_mutex.synchronize {
237
      self.relays[name] = relay
238
      self.rev_chans << channel
239
    }
240
    relay
241
  end
242

243
  #
244
  # Stop an active reverse port forward.
245
  #
246
  def stop_reverse_tcp_relay(rport)
247
    stop_relay("Reverse-#{rport}")
248
  end
249

250
  #
251
  # Starts a local TCP relay.
252
  #
253
  def start_tcp_relay(lport, opts = {})
254
    # Make sure our options are valid
255
    if ((opts['PeerHost'] == nil or opts['PeerPort'] == nil) and (opts['Stream'] != true))
256
      raise ArgumentError, "Missing peer host or peer port.", caller
257
    end
258

259
    listener = Rex::Socket.create_tcp_server(
260
      'LocalHost' => opts['LocalHost'],
261
      'LocalPort' => lport)
262

263
    _, lhost, lport = listener.getlocalname()
264
    opts['LocalHost']   = lhost
265
    opts['LocalPort']   = lport
266
    opts['__RelayType'] = 'tcp'
267

268
    start_relay(listener, lport.to_s + opts['LocalHost'], opts)
269
  end
270

271
  #
272
  # Starts a local relay on the supplied local port.  This listener will call
273
  # the supplied callback procedures when various events occur.
274
  #
275
  def start_relay(stream_server, name, opts = {})
276
    # Create a Relay instance with the local stream and remote stream
277
    relay = Relay.new(name, stream_server, opts)
278

279
    # Extend the stream_server so that we can associate it with this relay
280
    stream_server.extend(StreamServer)
281
    stream_server.relay = relay
282

283
    # Add the stream associations the appropriate lists and hashes
284
    self.relay_mutex.synchronize {
285
      self.relays[name] = relay
286

287
      self.rfds << stream_server
288
    }
289
    relay
290
  end
291

292
  #
293
  # Stops relaying on a given local port.
294
  #
295
  def stop_tcp_relay(lport, lhost = nil)
296
    stop_relay(lport.to_s + (lhost || '0.0.0.0'))
297
  end
298

299
  #
300
  # Stops a relay with a given name.
301
  #
302
  def stop_relay(name)
303
    rv = false
304

305
    self.relay_mutex.synchronize {
306
      relay = self.relays[name]
307

308
      if relay
309
        close_relay(relay)
310
        rv = true
311
      end
312
    }
313

314
    rv
315
  end
316

317
  #
318
  # Enumerate each TCP relay
319
  #
320
  def each_tcp_relay(&block)
321
    self.relays.each_pair { |name, relay|
322
      next if (relay.opts['__RelayType'] != 'tcp')
323

324
      yield(
325
        relay.opts['LocalHost'] || '0.0.0.0',
326
        relay.opts['LocalPort'],
327
        relay.opts['PeerHost'],
328
        relay.opts['PeerPort'],
329
        relay.opts)
330
    }
331
  end
332

333
protected
334

335
  attr_accessor :relays, :relay_thread, :relay_mutex
336
  attr_accessor :rfds, :rev_chans
337

338
  #
339
  # Closes an cleans up a specific relay
340
  #
341
  def close_relay(relay)
342

343
    if relay.kind_of?(ReverseRelay)
344
      self.rev_chans.delete(relay.channel)
345
    else
346
      self.rfds.delete(relay.listener)
347
    end
348

349
    self.relays.delete(relay.name)
350

351
    begin
352
      relay.shutdown
353
      relay.close
354
    rescue IOError
355
    end
356
  end
357

358
  #
359
  # Closes a specific relay connection without tearing down the actual relay
360
  # itself.
361
  #
362
  def close_relay_conn(fd)
363
    relay = fd.relay
364
    ofd   = fd.other_stream
365

366
    self.rfds.delete(fd)
367

368
    begin
369
      if relay.on_conn_close_proc
370
        relay.on_conn_close_proc.call(fd)
371
      end
372

373
      fd.shutdown
374
      fd.close
375
    rescue IOError
376
    end
377

378
    if ofd
379
      self.rfds.delete(ofd)
380

381
      begin
382
        if (relay.on_conn_close_proc)
383
          relay.on_conn_close_proc.call(ofd)
384
        end
385

386
        ofd.shutdown
387
        ofd.close
388
      rescue IOError
389
      end
390
    end
391
  end
392

393
  #
394
  # Attempt to accept a new reverse connection on the given reverse
395
  # relay handle.
396
  #
397
  def accept_reverse_relay(rrfd)
398

399
    rfd = rrfd.accept_nonblock
400

401
    return unless rfd
402

403
    lfd = Rex::Socket::Tcp.create(
404
      'PeerHost'  => rrfd.relay.opts['PeerHost'],
405
      'PeerPort'  => rrfd.relay.opts['PeerPort'],
406
      'Timeout'   => 5
407
    )
408

409
    rfd.extend(Stream)
410
    lfd.extend(Stream)
411

412
    rfd.relay = rrfd.relay
413
    lfd.relay = rrfd.relay
414

415
    self.rfds << lfd
416
    self.rfds << rfd
417

418
    rfd.other_stream = lfd
419
    lfd.other_stream = rfd
420
  end
421

422
  #
423
  # Accepts a client connection on a local relay.
424
  #
425
  def accept_relay_conn(srvfd)
426
    relay = srvfd.relay
427

428
    begin
429
      dlog("Accepting relay client connection...", 'rex', LEV_3)
430

431
      # Accept the child connection
432
      lfd = srvfd.accept
433
      dlog("Got left side of relay: #{lfd}", 'rex', LEV_3)
434

435
      # Call the relay's on_local_connection method which should return a
436
      # remote connection on success
437
      rfd = srvfd.on_local_connection(relay, lfd)
438

439
      dlog("Got right side of relay: #{rfd}", 'rex', LEV_3)
440
    rescue
441
      wlog("Failed to get remote half of local connection on relay #{relay.name}: #{$!}", 'rex')
442
      lfd.close
443
      return
444
    end
445

446
    # If we have both sides, then we rock.  Extend the instances, associate
447
    # them with the relay, associate them with each other, and add them to
448
    # the list of polling file descriptors
449
    if lfd && rfd
450
      lfd.extend(Stream)
451
      rfd.extend(Stream)
452

453
      lfd.relay = relay
454
      rfd.relay = relay
455

456
      lfd.other_stream = rfd
457
      rfd.other_stream = lfd
458

459
      self.rfds << lfd
460
      self.rfds << rfd
461
    else
462
      # Otherwise, we don't have both sides, we'll close them.
463
      close_relay_conn(lfd)
464
    end
465
  end
466

467
  #
468
  # Monitors the relays for data and passes it in both directions.
469
  #
470
  def monitor_relays
471
    begin
472
      # Helps with latency
473
      Thread.current.priority = 2
474

475
      # See if we have any new connections on the existing reverse port
476
      # forward relays
477
      rev_chans.each do |rrfd|
478
        accept_reverse_relay(rrfd)
479
      end
480

481
      # Poll all the streams...
482
      begin
483
        socks = Rex::ThreadSafe.select(rfds, nil, nil, 0.25)
484
      rescue StreamClosedError => e
485
        dlog("monitor_relays: closing stream #{e.stream}", 'rex', LEV_3)
486

487
        # Close the relay connection that is associated with the stream
488
        # closed error
489
        if e.stream.kind_of?(Stream)
490
          close_relay_conn(e.stream)
491
        end
492

493
        dlog("monitor_relays: closed stream #{e.stream}", 'rex', LEV_3)
494

495
        next
496
      rescue => e
497
        elog("Error in #{self} monitor_relays select:", 'rex', error: e)
498
        return
499
      end
500

501
      # If socks is nil, go again.
502
      next unless socks
503

504
      # Process read-ready file descriptors, if any.
505
      socks[0].each { |rfd|
506

507
        # If this file descriptor is a server, accept the connection
508
        if (rfd.kind_of?(StreamServer))
509
          accept_relay_conn(rfd)
510
        else
511
          # Otherwise, it's a relay connection, read data from one side
512
          # and write it to the other
513
          begin
514
            # Pass the data onto the other fd, most likely writing it.
515
            data = rfd.sysread(65536)
516
            rfd.other_stream.on_other_data(data)
517
          # If we catch an error, close the connection
518
          rescue ::Exception => e
519
            elog("Error in #{self} monitor_relays read", 'rex', error: e)
520
            close_relay_conn(rfd)
521
          end
522
        end
523

524
      } if (socks[0])
525

526
    end while true
527
  end
528

529
end
530

531
end
532
end
533

534

535