Class: Aikido::Zen::WorkerProcess::Agent::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/aikido/zen/worker_process/agent/client.rb

Constant Summary collapse

KEEPALIVE_INTERVAL =

The keepalive interval must be less than the RPC server read timeout.

4
ATTACK_WAVE_TIMEOUT =

This read timeout should be kept short because attackers can trigger attack wave detection checks at will, to limit how much latency an attacker can introduce per request.

1.0

Instance Method Summary collapse

Constructor Details

#initialize(host, port, secret: Aikido::Zen.secret, config: Aikido::Zen.config, worker: Aikido::Zen::Worker.new(config: config), keepalive_worker: Aikido::Zen::Worker.new(config: config), polling_interval: config.worker_process_polling_interval, polling_jitter: config.worker_process_polling_jitter, heartbeat_interval: config.worker_process_heartbeat_interval, collector: Aikido::Zen.collector) ⇒ Client

Returns a new instance of Client.



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
 
# File 'lib/aikido/zen/worker_process/agent/client.rb', line 16

def initialize(
  host,
  port,
  secret: Aikido::Zen.secret,
  config: Aikido::Zen.config,
  worker: Aikido::Zen::Worker.new(config: config),
  keepalive_worker: Aikido::Zen::Worker.new(config: config),
  polling_interval: config.worker_process_polling_interval,
  polling_jitter: config.worker_process_polling_jitter,
  heartbeat_interval: config.worker_process_heartbeat_interval,
  collector: Aikido::Zen.collector
)
  @config = config
  @worker = worker
  @keepalive_worker = keepalive_worker
  @polling_interval = polling_interval
  @polling_jitter = polling_jitter
  @heartbeat_interval = heartbeat_interval
  @collector = collector

  @rpc_client = Aikido::Zen::RPC::Client.new(secret, host, port, reconnect: true)

  @known_config_generation = nil
  @known_firewall_lists_generation = nil
end

Instance Method Details

#calculate_rate_limits(request) ⇒ Object 



64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/aikido/zen/worker_process/agent/client.rb', line 64

def calculate_rate_limits(request)
  result = @rpc_client.invoke(
    "calculate_rate_limits", Aikido::Zen::IPC::READ_TIMEOUT,
    request.route.as_json, request.client_ip, request.actor&.as_json
  )

  Aikido::Zen::RateLimiter::Result.from_json(result) if result
rescue => err
  @config.logger.error("Forked worker process #{Process.pid}: failed to get rate limits from parent: #{err.message}")
  raise
end

#close ⇒ Object 



42
43
44
 
# File 'lib/aikido/zen/worker_process/agent/client.rb', line 42

def close
  @rpc_client.close
end

#record_attack_wave(client_ip, sample) ⇒ Array<Aikido::Zen::AttackWave::Sample>?

Parameters:

Returns:



79
80
81
82
83
84
85
86
87
88
 
# File 'lib/aikido/zen/worker_process/agent/client.rb', line 79

def record_attack_wave(client_ip, sample)
  result = @rpc_client.invoke(
    "record_attack_wave", ATTACK_WAVE_TIMEOUT,
    client_ip, sample.verb, sample.path
  )
  result&.map { |data| Aikido::Zen::AttackWave::Sample.from_json(data) }
rescue => err
  @config.logger.error("Forked worker process #{Process.pid}: failed to record attack wave sample with parent: #{err.message}")
  raise
end

#send_collector_events ⇒ Object 



57
58
59
60
61
62
 
# File 'lib/aikido/zen/worker_process/agent/client.rb', line 57

def send_collector_events
  events_data = @collector.flush_events.map(&:as_json)
  @rpc_client.invoke("send_collector_events", KEEPALIVE_INTERVAL, events_data)
rescue => err
  @config.logger.error("Forked worker process #{Process.pid}: failed to send collector events to parent: #{err.message}")
end

#start ⇒ Object 



46
47
48
49
 
# File 'lib/aikido/zen/worker_process/agent/client.rb', line 46

def start
  @rpc_client.start
  schedule_tasks
end

#stop ⇒ Object 



51
52
53
54
55
 
# File 'lib/aikido/zen/worker_process/agent/client.rb', line 51

def stop
  @keepalive_worker.shutdown
  @worker.shutdown
  @rpc_client.stop
end