Class: Labimotion::AiEgressGuard

Inherits:
Object
  • Object
show all
Defined in:
lib/labimotion/libs/ai_egress_guard.rb

Overview

Validates a USER-SUPPLIED AI provider base URL before the server makes an outbound request to it, to prevent SSRF. A user may point the AI feature at their own OpenAI-compatible endpoint, but only at a PUBLIC https host — never at loopback / private / link-local / cloud-metadata addresses.

Known limitation: validation resolves DNS at check time; a hostile resolver could rebind between this check and the actual request (TOCTOU). Redirect following is disabled at the HTTP layer, which removes the most common bypass; pinning the connection to the validated IP would close the residual window.

Defined Under Namespace

Classes: BlockedError

Constant Summary collapse

EXTRA_BLOCKED =

Ranges Ruby's IPAddr#private?/#loopback?/#link_local? don't all cover: unspecified, CGNAT, benchmarking/test-net, IPv4-broadcast, reserved, and the IPv6 unique-local / link-local / multicast blocks.

[
  '0.0.0.0/8', '100.64.0.0/10', '192.0.0.0/24', '192.0.2.0/24',
  '198.18.0.0/15', '198.51.100.0/24', '203.0.113.0/24', '240.0.0.0/4',
  '255.255.255.255/32', '::/128', '::1/128', 'fc00::/7', 'fe80::/10', 'ff00::/8'
].map { |cidr| IPAddr.new(cidr) }.freeze

Class Method Summary collapse

Class Method Details

.allowed_ip?(ip) ⇒ Boolean

Outside production, a private-network address (e.g. a self-hosted model on another machine on the LAN) is allowed; loopback/link-local/other EXTRA_BLOCKED ranges (cloud metadata etc.) stay blocked regardless of env.

Returns:

  • (Boolean) 


90
91
92
93
94
95
96
97
 
# File 'lib/labimotion/libs/ai_egress_guard.rb', line 90

def self.allowed_ip?(ip)
  return public_ip?(ip) if Rails.env.production?
  return false if ip.loopback? || ip.link_local?

  EXTRA_BLOCKED.none? { |net| net.include?(ip) }
rescue StandardError
  false
end

.literal_ip?(str) ⇒ Boolean

Returns:

  • (Boolean) 


64
65
66
67
68
69
 
# File 'lib/labimotion/libs/ai_egress_guard.rb', line 64

def self.literal_ip?(str)
  IPAddr.new(str)
  true
rescue IPAddr::InvalidAddressError
  false
end

.parse(url) ⇒ Object 



49
50
51
52
53
 
# File 'lib/labimotion/libs/ai_egress_guard.rb', line 49

def self.parse(url)
  URI.parse(url.to_s.strip)
rescue URI::InvalidURIError
  raise BlockedError, 'AI provider URL is not a valid URL'
end

.public_ip?(ip) ⇒ Boolean

Public = not loopback/private/link-local and not in EXTRA_BLOCKED.

Returns:

  • (Boolean) 


79
80
81
82
83
84
85
 
# File 'lib/labimotion/libs/ai_egress_guard.rb', line 79

def self.public_ip?(ip)
  return false if ip.loopback? || ip.private? || ip.link_local?

  EXTRA_BLOCKED.none? { |net| net.include?(ip) }
rescue StandardError
  false
end

.resolve(host) ⇒ Object

All A/AAAA addresses for host as IPAddr; a literal IP resolves to itself.



56
57
58
59
60
61
62
 
# File 'lib/labimotion/libs/ai_egress_guard.rb', line 56

def self.resolve(host)
  return [IPAddr.new(host)] if literal_ip?(host)

  Resolv.getaddresses(host).filter_map { |addr| safe_ipaddr(addr) }
rescue StandardError
  []
end

.safe_ipaddr(addr) ⇒ Object 



71
72
73
74
75
76
 
# File 'lib/labimotion/libs/ai_egress_guard.rb', line 71

def self.safe_ipaddr(addr)
  ip = IPAddr.new(addr.to_s)
  ip.ipv4_mapped? ? ip.native : ip
rescue StandardError
  nil
end

.validate!(url) ⇒ Object

Raise BlockedError unless url is a public https endpoint (http is also accepted outside production, e.g. against a local dev provider). Returns url.

Raises:



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/labimotion/libs/ai_egress_guard.rb', line 31

def self.validate!(url)
  uri = parse(url)
  allowed_schemes = Rails.env.production? ? %w[https] : %w[https http]
  raise BlockedError, "AI provider URL must use #{allowed_schemes.join(' or ')}" unless
    allowed_schemes.include?(uri.scheme)

  host = uri.host.to_s
  raise BlockedError, 'AI provider URL has no host' if host.empty?

  addresses = resolve(host)
  raise BlockedError, "AI provider host could not be resolved: #{host}" if addresses.empty?

  blocked = addresses.find { |ip| !allowed_ip?(ip) }
  raise BlockedError, "AI provider host resolves to a non-public address (#{blocked})" if blocked

  url
end