Phase 1: Foundation Types (100% complete)
- query_helpers: SQL LIKE sanitization with pipe operators
- numeric: Integer parsing with pattern matching guards
- result: Pure Elixir Result monad (map, and_then, unwrap_or)
Phase 2: Ecto Domain Types (100% complete)
- ip_address: IPv4/IPv6 validation using :inet directly
- mac_address: Multi-format MAC parsing (colon/hyphen/dot/compact)
- snmp_oid: OID parsing/manipulation with recursive pattern matching
All 198 tests passing across converted modules.
API changed from Gleam-style {:some/:none to idiomatic {:ok/:error.
Refactored parse_numeric_oid to use with statement, reducing nesting depth.
Reviewed-on: graham/towerops-web#196
126 lines
3.7 KiB
Elixir
126 lines
3.7 KiB
Elixir
defmodule Towerops.URLValidator do
|
|
@moduledoc """
|
|
Validates URLs are safe for server-side requests (prevents SSRF).
|
|
|
|
Blocks:
|
|
- Non-HTTP(S) schemes
|
|
- Localhost and .local domains
|
|
- Private/internal IP addresses (127.0.0.0/8, 10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16, 169.254.0.0/16)
|
|
- IPv6 loopback (::1)
|
|
"""
|
|
|
|
@doc """
|
|
Validates a URL is safe for server-side requests.
|
|
|
|
Returns `:ok` or `{:error, reason}`.
|
|
|
|
## Examples
|
|
|
|
iex> URLValidator.validate("https://example.com")
|
|
:ok
|
|
|
|
iex> URLValidator.validate("http://localhost")
|
|
{:error, "Requests to localhost are not allowed"}
|
|
|
|
iex> URLValidator.validate("http://192.168.1.1")
|
|
{:error, "Requests to private/internal IP addresses are not allowed"}
|
|
"""
|
|
@spec validate(term()) :: :ok | {:error, String.t()}
|
|
def validate(value) when is_binary(value) do
|
|
with {:ok, scheme, host} <- parse_url(value),
|
|
:ok <- validate_scheme(scheme),
|
|
:ok <- validate_host(host) do
|
|
validate_ip(host)
|
|
end
|
|
end
|
|
|
|
def validate(_), do: {:error, "URL must be a string"}
|
|
|
|
# Parse URL into scheme and host
|
|
defp parse_url(url) do
|
|
uri = URI.parse(url)
|
|
|
|
with scheme when not is_nil(scheme) <- uri.scheme,
|
|
host when not is_nil(host) and host != "" <- uri.host do
|
|
{:ok, scheme, host}
|
|
else
|
|
nil -> {:error, "URL must include a scheme (http or https)"}
|
|
"" -> {:error, "URL must include a host"}
|
|
end
|
|
end
|
|
|
|
# Validate scheme is HTTP or HTTPS
|
|
defp validate_scheme(scheme) when scheme in ["http", "https"], do: :ok
|
|
defp validate_scheme(_), do: {:error, "Only http and https schemes are allowed"}
|
|
|
|
# Validate host is not localhost or .local
|
|
defp validate_host(host) do
|
|
downcased = String.downcase(host)
|
|
|
|
if downcased == "localhost" or String.ends_with?(downcased, ".local") do
|
|
{:error, "Requests to localhost are not allowed"}
|
|
else
|
|
:ok
|
|
end
|
|
end
|
|
|
|
# Validate host doesn't resolve to private IP
|
|
defp validate_ip(host) do
|
|
case resolve_host(host) do
|
|
{:ok, ip_tuple} ->
|
|
if private_ip?(ip_tuple) do
|
|
{:error, "Requests to private/internal IP addresses are not allowed"}
|
|
else
|
|
:ok
|
|
end
|
|
|
|
# Can't resolve — let the HTTP client handle DNS errors
|
|
{:error, _} ->
|
|
:ok
|
|
end
|
|
end
|
|
|
|
# Resolve hostname to IP address
|
|
defp resolve_host(host) do
|
|
charlist = String.to_charlist(host)
|
|
|
|
# Try parsing as IP first
|
|
case :inet.parse_address(charlist) do
|
|
{:ok, ip} ->
|
|
{:ok, ip}
|
|
|
|
{:error, _} ->
|
|
# Try DNS resolution (IPv4 first, then IPv6)
|
|
case :inet.getaddr(charlist, :inet) do
|
|
{:ok, ip} -> {:ok, ip}
|
|
{:error, _} -> :inet.getaddr(charlist, :inet6)
|
|
end
|
|
end
|
|
end
|
|
|
|
# Check if IP is private/internal
|
|
defp private_ip?({0, 0, 0, 0, 0, 0, 0, 1}), do: true
|
|
|
|
defp private_ip?({a, b, c, d}) when tuple_size({a, b, c, d}) == 4 do
|
|
ip_int = Bitwise.bsl(a, 24) + Bitwise.bsl(b, 16) + Bitwise.bsl(c, 8) + d
|
|
|
|
# 127.0.0.0/8 (loopback)
|
|
# 10.0.0.0/8 (private)
|
|
# 172.16.0.0/12 (private)
|
|
# 192.168.0.0/16 (private)
|
|
# 169.254.0.0/16 (link-local)
|
|
in_cidr?(ip_int, Bitwise.bsl(127, 24), 8) or
|
|
in_cidr?(ip_int, Bitwise.bsl(10, 24), 8) or
|
|
in_cidr?(ip_int, Bitwise.bsl(172, 24) + Bitwise.bsl(16, 16), 12) or
|
|
in_cidr?(ip_int, Bitwise.bsl(192, 24) + Bitwise.bsl(168, 16), 16) or
|
|
in_cidr?(ip_int, Bitwise.bsl(169, 24) + Bitwise.bsl(254, 16), 16)
|
|
end
|
|
|
|
defp private_ip?(_), do: false
|
|
|
|
# Check if IP is in CIDR range
|
|
defp in_cidr?(ip_int, net_int, prefix) do
|
|
mask = Bitwise.band(Bitwise.bsl(0xFFFFFFFF, 32 - prefix), 0xFFFFFFFF)
|
|
Bitwise.band(ip_int, mask) == Bitwise.band(net_int, mask)
|
|
end
|
|
end
|