Graham McIntire
44fc017653
- Update pre-commit config - Add test coverage for URL validation, security modules, and monitoring executors - Improve network map implementation and tests - Add coverage for API controllers and LiveView endpoints - Refactor code to fix Credo issues (reduce nesting, extract helper functions)
104 lines
2.7 KiB
Elixir
104 lines
2.7 KiB
Elixir
defmodule Towerops.URLValidator do
|
|
@moduledoc """
|
|
Validates URLs to prevent SSRF (Server-Side Request Forgery) attacks.
|
|
|
|
Blocks requests to internal/private IP ranges, localhost, and non-HTTP(S) schemes.
|
|
"""
|
|
|
|
import Bitwise
|
|
|
|
@private_ranges [
|
|
# 127.0.0.0/8
|
|
{127, 0, 0, 0, 8},
|
|
# 10.0.0.0/8
|
|
{10, 0, 0, 0, 8},
|
|
# 172.16.0.0/12
|
|
{172, 16, 0, 0, 12},
|
|
# 192.168.0.0/16
|
|
{192, 168, 0, 0, 16},
|
|
# 169.254.0.0/16 (link-local)
|
|
{169, 254, 0, 0, 16},
|
|
# IPv6 loopback ::1
|
|
:ipv6_loopback
|
|
]
|
|
|
|
@doc """
|
|
Validates a URL is safe for server-side requests.
|
|
|
|
Returns :ok or {:error, reason}.
|
|
"""
|
|
@spec validate(String.t()) :: :ok | {:error, String.t()}
|
|
def validate(url) when is_binary(url) do
|
|
with {:ok, uri} <- parse_url(url),
|
|
:ok <- validate_scheme(uri),
|
|
:ok <- validate_host(uri) do
|
|
validate_ip(uri.host)
|
|
end
|
|
end
|
|
|
|
def validate(_), do: {:error, "URL must be a string"}
|
|
|
|
defp parse_url(url) do
|
|
case URI.parse(url) do
|
|
%URI{scheme: nil} -> {:error, "URL must include a scheme (http or https)"}
|
|
%URI{host: nil} -> {:error, "URL must include a host"}
|
|
%URI{host: ""} -> {:error, "URL must include a host"}
|
|
uri -> {:ok, uri}
|
|
end
|
|
end
|
|
|
|
defp validate_scheme(%URI{scheme: scheme}) when scheme in ["http", "https"], do: :ok
|
|
defp validate_scheme(_), do: {:error, "Only http and https schemes are allowed"}
|
|
|
|
defp validate_host(%URI{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
|
|
|
|
defp validate_ip(host) do
|
|
case resolve_host(host) do
|
|
{:ok, ip} ->
|
|
if private_ip?(ip) 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
|
|
|
|
defp resolve_host(host) do
|
|
charlist = String.to_charlist(host)
|
|
|
|
with {:error, _} <- :inet.parse_address(charlist),
|
|
{:error, _} <- :inet.getaddr(charlist, :inet) do
|
|
error = :inet.getaddr(charlist, :inet6)
|
|
error
|
|
end
|
|
end
|
|
|
|
defp private_ip?({0, 0, 0, 0, 0, 0, 0, 1}), do: true
|
|
|
|
defp private_ip?({a, b, c, d}) do
|
|
Enum.any?(@private_ranges, fn
|
|
:ipv6_loopback ->
|
|
false
|
|
|
|
{net_a, net_b, net_c, net_d, prefix} ->
|
|
mask = 0xFFFFFFFF |> bsl(32 - prefix) |> band(0xFFFFFFFF)
|
|
ip_int = bsl(a, 24) + bsl(b, 16) + bsl(c, 8) + d
|
|
net_int = bsl(net_a, 24) + bsl(net_b, 16) + bsl(net_c, 8) + net_d
|
|
band(ip_int, mask) == band(net_int, mask)
|
|
end)
|
|
end
|
|
|
|
defp private_ip?(_), do: false
|
|
end
|