Graham McIntire
f2cd3488ee
the version regex only allowed [a-zA-Z0-9.] in semver suffixes, which rejected hyphens from git-describe output (e.g. 0.1.0-5-gabcdef0). this caused every heartbeat to fail validation, preventing last_seen_at from updating, which led to agent status showing "warning" and eventual channel disconnect after 360s. also accept bare git SHAs and short identifiers like "dev".
538 lines
19 KiB
Elixir
538 lines
19 KiB
Elixir
defmodule Towerops.Agent.Validator do
|
|
@moduledoc """
|
|
Strict validation for Protocol Buffer messages from agents.
|
|
|
|
Provides validation on top of protobuf schema to catch malformed data,
|
|
injection attempts, and business logic violations before processing.
|
|
|
|
## Philosophy
|
|
|
|
Protobuf provides schema-level validation (types, required fields) but
|
|
doesn't validate business constraints. This module adds:
|
|
|
|
- String length limits (prevent memory exhaustion)
|
|
- Format validation (IP addresses, OIDs, UUIDs)
|
|
- Range validation (ports, intervals, counters)
|
|
- Enum validation (strict whitelisting)
|
|
- Content validation (prevent injection)
|
|
|
|
## Usage
|
|
|
|
# Validate heartbeat
|
|
{:ok, heartbeat} = Validator.validate_heartbeat(binary)
|
|
|
|
# Validate metric batch
|
|
{:ok, metrics} = Validator.validate_metric_batch(binary)
|
|
|
|
# Validate SNMP result
|
|
{:ok, result} = Validator.validate_snmp_result(binary)
|
|
"""
|
|
|
|
alias Towerops.Agent.AgentError
|
|
alias Towerops.Agent.AgentHeartbeat
|
|
alias Towerops.Agent.CredentialTestResult
|
|
alias Towerops.Agent.InterfaceStat
|
|
alias Towerops.Agent.Metric
|
|
alias Towerops.Agent.MetricBatch
|
|
alias Towerops.Agent.MikrotikResult
|
|
alias Towerops.Agent.MonitoringCheck
|
|
alias Towerops.Agent.NeighborDiscovery
|
|
alias Towerops.Agent.SensorReading
|
|
alias Towerops.Agent.SnmpResult
|
|
|
|
require Logger
|
|
|
|
# String length limits (prevent memory exhaustion attacks)
|
|
@max_string_length 10_000
|
|
@max_short_string_length 255
|
|
@max_error_message_length 1_000
|
|
|
|
# Numeric limits (reserved for future use)
|
|
# @max_oid_length 128
|
|
# @max_port 65_535
|
|
# @max_interval_seconds 86_400
|
|
# uint64 max (136 years)
|
|
@max_uptime_seconds 4_294_967_295
|
|
# 1 minute
|
|
@max_response_time_ms 60_000
|
|
# 2038-01-19 (int32 limit)
|
|
@max_timestamp 2_147_483_647
|
|
|
|
# Collection limits (prevent DoS via large batches)
|
|
@max_metrics_per_batch 10_000
|
|
# Discovery on enterprise devices can return 10k-50k OIDs (interfaces, routing tables, etc.)
|
|
@max_oid_values 50_000
|
|
@max_mikrotik_sentences 1_000
|
|
@max_capabilities 20
|
|
|
|
# Valid enums
|
|
@valid_statuses ["success", "failure", "up", "down", "ok", "error", "warning"]
|
|
@valid_protocols ["lldp", "cdp"]
|
|
|
|
@type validation_error :: {:error, {atom(), binary()}}
|
|
@type validation_result(t) :: {:ok, t} | validation_error()
|
|
|
|
## Public API - Message Validation
|
|
|
|
@doc """
|
|
Validate AgentHeartbeat message.
|
|
|
|
Checks version format, hostname validity, and uptime range.
|
|
"""
|
|
@spec validate_heartbeat(binary()) :: validation_result(AgentHeartbeat.t())
|
|
def validate_heartbeat(binary) when is_binary(binary) do
|
|
with {:ok, heartbeat} <- safe_decode(AgentHeartbeat, binary),
|
|
:ok <- validate_version(heartbeat.version),
|
|
:ok <- validate_hostname(heartbeat.hostname),
|
|
:ok <- validate_uptime(heartbeat.uptime_seconds),
|
|
:ok <- validate_optional_ip(heartbeat.ip_address) do
|
|
{:ok, heartbeat}
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Validate MetricBatch message.
|
|
|
|
Checks batch size limits and validates each metric individually.
|
|
"""
|
|
@spec validate_metric_batch(binary()) :: validation_result(MetricBatch.t())
|
|
def validate_metric_batch(binary) when is_binary(binary) do
|
|
with {:ok, batch} <- safe_decode(MetricBatch, binary),
|
|
:ok <- validate_batch_size(batch.metrics),
|
|
:ok <- validate_all_metrics(batch.metrics) do
|
|
{:ok, batch}
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Validate SnmpResult message.
|
|
|
|
Checks device_id format, OID values, and timestamp.
|
|
"""
|
|
@spec validate_snmp_result(binary()) :: validation_result(SnmpResult.t())
|
|
def validate_snmp_result(binary) when is_binary(binary) do
|
|
with {:ok, result} <- safe_decode(SnmpResult, binary),
|
|
:ok <- validate_device_id(result.device_id),
|
|
:ok <- validate_oid_values(result.oid_values),
|
|
:ok <- validate_timestamp(result.timestamp) do
|
|
{:ok, result}
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Validate AgentError message.
|
|
|
|
Checks device_id, job_id, and error message length.
|
|
"""
|
|
@spec validate_agent_error(binary()) :: validation_result(AgentError.t())
|
|
def validate_agent_error(binary) when is_binary(binary) do
|
|
with {:ok, error} <- safe_decode(AgentError, binary),
|
|
:ok <- validate_device_id(error.device_id),
|
|
:ok <- validate_job_id(error.job_id),
|
|
:ok <- validate_error_message(error.message) do
|
|
{:ok, error}
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Validate CredentialTestResult message.
|
|
|
|
Checks test_id, error message, and system description.
|
|
"""
|
|
@spec validate_credential_test_result(binary()) :: validation_result(CredentialTestResult.t())
|
|
def validate_credential_test_result(binary) when is_binary(binary) do
|
|
with {:ok, result} <- safe_decode(CredentialTestResult, binary),
|
|
:ok <- validate_test_id(result.test_id),
|
|
:ok <- validate_error_message(result.error_message),
|
|
:ok <- validate_system_description(result.system_description),
|
|
:ok <- validate_timestamp(result.timestamp) do
|
|
{:ok, result}
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Validate MikrotikResult message.
|
|
|
|
Checks device_id, job_id, sentences count, and error message.
|
|
"""
|
|
@spec validate_mikrotik_result(binary()) :: validation_result(MikrotikResult.t())
|
|
def validate_mikrotik_result(binary) when is_binary(binary) do
|
|
with {:ok, result} <- safe_decode(MikrotikResult, binary),
|
|
:ok <- validate_device_id(result.device_id),
|
|
:ok <- validate_job_id(result.job_id),
|
|
:ok <- validate_mikrotik_sentences(result.sentences),
|
|
:ok <- validate_error_message(result.error),
|
|
:ok <- validate_timestamp(result.timestamp) do
|
|
{:ok, result}
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Validate MonitoringCheck message from binary protobuf.
|
|
|
|
Decodes the binary and validates device_id, status enum, response_time range, and timestamp.
|
|
"""
|
|
@spec validate_monitoring_check_message(binary()) :: validation_result(MonitoringCheck.t())
|
|
def validate_monitoring_check_message(binary) when is_binary(binary) do
|
|
with {:ok, check} <- safe_decode(MonitoringCheck, binary),
|
|
:ok <- validate_monitoring_check(check) do
|
|
{:ok, check}
|
|
end
|
|
end
|
|
|
|
## Private Validation Functions
|
|
|
|
# Safe decode with error handling
|
|
defp safe_decode(module, binary) do
|
|
{:ok, module.decode(binary)}
|
|
rescue
|
|
e in [Protobuf.DecodeError, ArgumentError] ->
|
|
Logger.error("Protobuf decode error: #{Exception.message(e)}")
|
|
{:error, {:decode_error, "Invalid protobuf message format"}}
|
|
end
|
|
|
|
# Validate version format - accepts semver, RFC 3339 timestamp, git describe,
|
|
# git short SHA, or "dev". Empty string allowed for backward compat.
|
|
defp validate_version(""), do: :ok
|
|
|
|
defp validate_version(version) when is_binary(version) do
|
|
cond do
|
|
String.length(version) > @max_short_string_length ->
|
|
{:error, {:invalid_version, "Version too long"}}
|
|
|
|
# Semantic version with git-describe suffixes (e.g., 1.2.3, 1.2.3-rc.1, 1.2.3-5-gabcdef0)
|
|
String.match?(version, ~r/^\d+\.\d+\.\d+(-[a-zA-Z0-9.\-]+)?$/) ->
|
|
:ok
|
|
|
|
# RFC 3339 timestamp format (e.g., 2026-02-11T17:00:00Z)
|
|
String.match?(version, ~r/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$/) ->
|
|
:ok
|
|
|
|
# Git short SHA (7-40 hex chars) or short identifier like "dev"
|
|
String.match?(version, ~r/^[a-zA-Z0-9]{1,40}$/) ->
|
|
:ok
|
|
|
|
true ->
|
|
{:error, {:invalid_version, "Version must be semver, RFC 3339 timestamp, or short identifier"}}
|
|
end
|
|
end
|
|
|
|
defp validate_version(_), do: {:error, {:invalid_version, "Version must be a string"}}
|
|
|
|
# Validate hostname (RFC 1123, empty string allowed for backward compat)
|
|
defp validate_hostname(""), do: :ok
|
|
|
|
defp validate_hostname(hostname) when is_binary(hostname) do
|
|
cond do
|
|
String.length(hostname) > @max_short_string_length ->
|
|
{:error, {:invalid_hostname, "Hostname too long"}}
|
|
|
|
String.match?(
|
|
hostname,
|
|
~r/^[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(\.[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)*$/
|
|
) ->
|
|
:ok
|
|
|
|
true ->
|
|
{:error, {:invalid_hostname, "Hostname must be valid RFC 1123 format"}}
|
|
end
|
|
end
|
|
|
|
defp validate_hostname(""), do: :ok
|
|
defp validate_hostname(_), do: {:error, {:invalid_hostname, "Hostname must be a string"}}
|
|
|
|
# Validate uptime seconds
|
|
defp validate_uptime(uptime) when is_integer(uptime) and uptime >= 0 and uptime <= @max_uptime_seconds do
|
|
:ok
|
|
end
|
|
|
|
defp validate_uptime(_), do: {:error, {:invalid_uptime, "Uptime must be 0 to #{@max_uptime_seconds}"}}
|
|
|
|
# Validate optional IP address
|
|
defp validate_optional_ip(""), do: :ok
|
|
defp validate_optional_ip(nil), do: :ok
|
|
|
|
defp validate_optional_ip(ip) when is_binary(ip) do
|
|
case :inet.parse_address(String.to_charlist(ip)) do
|
|
{:ok, _} -> :ok
|
|
{:error, _} -> {:error, {:invalid_ip, "IP address must be valid IPv4 or IPv6"}}
|
|
end
|
|
end
|
|
|
|
# Validate batch size
|
|
defp validate_batch_size(metrics) when is_list(metrics) do
|
|
if length(metrics) <= @max_metrics_per_batch do
|
|
:ok
|
|
else
|
|
{:error, {:batch_too_large, "Batch cannot exceed #{@max_metrics_per_batch} metrics"}}
|
|
end
|
|
end
|
|
|
|
# Validate all metrics in batch
|
|
defp validate_all_metrics(metrics) do
|
|
Enum.reduce_while(metrics, :ok, fn metric, _acc ->
|
|
case validate_metric(metric) do
|
|
:ok -> {:cont, :ok}
|
|
error -> {:halt, error}
|
|
end
|
|
end)
|
|
end
|
|
|
|
# Validate individual metric
|
|
defp validate_metric(%Metric{metric_type: {:sensor_reading, reading}}) do
|
|
validate_sensor_reading(reading)
|
|
end
|
|
|
|
defp validate_metric(%Metric{metric_type: {:interface_stat, stat}}) do
|
|
validate_interface_stat(stat)
|
|
end
|
|
|
|
defp validate_metric(%Metric{metric_type: {:neighbor_discovery, discovery}}) do
|
|
validate_neighbor_discovery(discovery)
|
|
end
|
|
|
|
defp validate_metric(%Metric{metric_type: {:monitoring_check, check}}) do
|
|
validate_monitoring_check(check)
|
|
end
|
|
|
|
defp validate_metric(_), do: {:error, {:invalid_metric, "Unknown metric type"}}
|
|
|
|
# Validate SensorReading
|
|
defp validate_sensor_reading(%SensorReading{} = reading) do
|
|
with :ok <- validate_sensor_id(reading.sensor_id),
|
|
:ok <- validate_sensor_value(reading.value),
|
|
:ok <- validate_status(reading.status) do
|
|
validate_timestamp(reading.timestamp)
|
|
end
|
|
end
|
|
|
|
# Validate InterfaceStat
|
|
defp validate_interface_stat(%InterfaceStat{} = stat) do
|
|
with :ok <- validate_interface_id(stat.interface_id),
|
|
:ok <- validate_counter(stat.if_in_octets, "if_in_octets"),
|
|
:ok <- validate_counter(stat.if_out_octets, "if_out_octets"),
|
|
:ok <- validate_counter(stat.if_in_errors, "if_in_errors"),
|
|
:ok <- validate_counter(stat.if_out_errors, "if_out_errors"),
|
|
:ok <- validate_counter(stat.if_in_discards, "if_in_discards"),
|
|
:ok <- validate_counter(stat.if_out_discards, "if_out_discards") do
|
|
validate_timestamp(stat.timestamp)
|
|
end
|
|
end
|
|
|
|
# Validate NeighborDiscovery
|
|
defp validate_neighbor_discovery(%NeighborDiscovery{} = discovery) do
|
|
with :ok <- validate_interface_id(discovery.interface_id),
|
|
:ok <- validate_protocol(discovery.protocol),
|
|
:ok <- validate_short_string(discovery.remote_chassis_id, "remote_chassis_id"),
|
|
:ok <- validate_short_string(discovery.remote_system_name, "remote_system_name"),
|
|
:ok <- validate_string(discovery.remote_system_description, "remote_system_description"),
|
|
:ok <- validate_short_string(discovery.remote_platform, "remote_platform"),
|
|
:ok <- validate_short_string(discovery.remote_port_id, "remote_port_id"),
|
|
:ok <- validate_short_string(discovery.remote_port_description, "remote_port_description"),
|
|
:ok <- validate_optional_ip(discovery.remote_address),
|
|
:ok <- validate_capabilities(discovery.remote_capabilities) do
|
|
validate_timestamp(discovery.timestamp)
|
|
end
|
|
end
|
|
|
|
# Validate MonitoringCheck
|
|
defp validate_monitoring_check(%MonitoringCheck{} = check) do
|
|
with :ok <- validate_device_id(check.device_id),
|
|
:ok <- validate_status(check.status),
|
|
:ok <- validate_response_time(check.response_time_ms) do
|
|
validate_timestamp(check.timestamp)
|
|
end
|
|
end
|
|
|
|
# Validate device ID (UUID format)
|
|
defp validate_device_id(id) when is_binary(id) do
|
|
if String.match?(id, ~r/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i) do
|
|
:ok
|
|
else
|
|
{:error, {:invalid_device_id, "Device ID must be valid UUID"}}
|
|
end
|
|
end
|
|
|
|
defp validate_device_id(_), do: {:error, {:invalid_device_id, "Device ID must be a string"}}
|
|
|
|
# Validate sensor ID
|
|
defp validate_sensor_id(id) when is_binary(id) do
|
|
if String.match?(id, ~r/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i) do
|
|
:ok
|
|
else
|
|
{:error, {:invalid_sensor_id, "Sensor ID must be valid UUID"}}
|
|
end
|
|
end
|
|
|
|
defp validate_sensor_id(_), do: {:error, {:invalid_sensor_id, "Sensor ID must be a string"}}
|
|
|
|
# Validate interface ID
|
|
defp validate_interface_id(id) when is_binary(id) do
|
|
if String.match?(id, ~r/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i) do
|
|
:ok
|
|
else
|
|
{:error, {:invalid_interface_id, "Interface ID must be valid UUID"}}
|
|
end
|
|
end
|
|
|
|
defp validate_interface_id(_), do: {:error, {:invalid_interface_id, "Interface ID must be a string"}}
|
|
|
|
# Validate test ID
|
|
defp validate_test_id(id) when is_binary(id) do
|
|
if String.match?(id, ~r/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i) do
|
|
:ok
|
|
else
|
|
{:error, {:invalid_test_id, "Test ID must be valid UUID"}}
|
|
end
|
|
end
|
|
|
|
defp validate_test_id(_), do: {:error, {:invalid_test_id, "Test ID must be a string"}}
|
|
|
|
# Validate job ID
|
|
defp validate_job_id(id) when is_binary(id) and byte_size(id) > 0 and byte_size(id) <= @max_short_string_length do
|
|
:ok
|
|
end
|
|
|
|
defp validate_job_id(_), do: {:error, {:invalid_job_id, "Job ID must be non-empty string"}}
|
|
|
|
# Validate sensor value (must be finite float/int)
|
|
defp validate_sensor_value(value) when is_integer(value), do: :ok
|
|
|
|
defp validate_sensor_value(value) when is_float(value) do
|
|
# Check for NaN (NaN is the only value that doesn't equal itself)
|
|
# Check for infinity by comparing with max/min float values
|
|
cond do
|
|
# credo:disable-for-next-line Credo.Check.Warning.OperationOnSameValues
|
|
value != value ->
|
|
{:error, {:invalid_sensor_value, "Sensor value must be finite number (not NaN)"}}
|
|
|
|
value > 1.0e308 or value < -1.0e308 ->
|
|
{:error, {:invalid_sensor_value, "Sensor value must be finite number (not infinity)"}}
|
|
|
|
true ->
|
|
:ok
|
|
end
|
|
end
|
|
|
|
defp validate_sensor_value(_), do: {:error, {:invalid_sensor_value, "Sensor value must be number"}}
|
|
|
|
# Validate counter (must be non-negative int64)
|
|
defp validate_counter(value, _field) when is_integer(value) and value >= 0 do
|
|
:ok
|
|
end
|
|
|
|
defp validate_counter(_, _field), do: {:error, {:invalid_counter, "Counter must be non-negative integer"}}
|
|
|
|
# Validate status enum
|
|
defp validate_status(status) when status in @valid_statuses, do: :ok
|
|
defp validate_status(""), do: :ok
|
|
defp validate_status(status), do: {:error, {:invalid_status, "Status '#{status}' not in #{inspect(@valid_statuses)}"}}
|
|
|
|
# Validate protocol enum
|
|
defp validate_protocol(protocol) when protocol in @valid_protocols, do: :ok
|
|
|
|
defp validate_protocol(protocol),
|
|
do: {:error, {:invalid_protocol, "Protocol '#{protocol}' not in #{inspect(@valid_protocols)}"}}
|
|
|
|
# Validate response time
|
|
defp validate_response_time(ms) when is_float(ms) and ms >= 0.0 and ms <= @max_response_time_ms do
|
|
:ok
|
|
end
|
|
|
|
defp validate_response_time(_),
|
|
do: {:error, {:invalid_response_time, "Response time must be 0 to #{@max_response_time_ms}ms"}}
|
|
|
|
# Validate timestamp (Unix epoch seconds)
|
|
defp validate_timestamp(ts) when is_integer(ts) and ts >= 0 and ts <= @max_timestamp do
|
|
:ok
|
|
end
|
|
|
|
defp validate_timestamp(_),
|
|
do: {:error, {:invalid_timestamp, "Timestamp must be valid Unix epoch (0 to #{@max_timestamp})"}}
|
|
|
|
# Validate short string (hostnames, IDs, etc.)
|
|
defp validate_short_string(str, field) when is_binary(str) do
|
|
if String.length(str) <= @max_short_string_length do
|
|
:ok
|
|
else
|
|
{:error, {:string_too_long, "#{field} exceeds #{@max_short_string_length} characters"}}
|
|
end
|
|
end
|
|
|
|
defp validate_short_string("", _field), do: :ok
|
|
defp validate_short_string(_, field), do: {:error, {:invalid_string, "#{field} must be a string"}}
|
|
|
|
# Validate general string (descriptions, etc.)
|
|
defp validate_string(str, field) when is_binary(str) do
|
|
if String.length(str) <= @max_string_length do
|
|
:ok
|
|
else
|
|
{:error, {:string_too_long, "#{field} exceeds #{@max_string_length} characters"}}
|
|
end
|
|
end
|
|
|
|
defp validate_string("", _field), do: :ok
|
|
defp validate_string(_, field), do: {:error, {:invalid_string, "#{field} must be a string"}}
|
|
|
|
# Validate error message
|
|
defp validate_error_message(msg) when is_binary(msg) do
|
|
if String.length(msg) <= @max_error_message_length do
|
|
:ok
|
|
else
|
|
{:error, {:string_too_long, "Error message exceeds #{@max_error_message_length} characters"}}
|
|
end
|
|
end
|
|
|
|
defp validate_error_message(""), do: :ok
|
|
defp validate_error_message(_), do: {:error, {:invalid_error_message, "Error message must be a string"}}
|
|
|
|
# Validate system description
|
|
defp validate_system_description(desc) when is_binary(desc) do
|
|
if String.length(desc) <= @max_string_length do
|
|
:ok
|
|
else
|
|
{:error, {:string_too_long, "System description exceeds #{@max_string_length} characters"}}
|
|
end
|
|
end
|
|
|
|
defp validate_system_description(""), do: :ok
|
|
defp validate_system_description(_), do: {:error, {:invalid_system_description, "System description must be a string"}}
|
|
|
|
# Validate capabilities list
|
|
defp validate_capabilities(caps) when is_list(caps) do
|
|
cond do
|
|
length(caps) > @max_capabilities ->
|
|
{:error, {:too_many_capabilities, "Capabilities list exceeds #{@max_capabilities} items"}}
|
|
|
|
Enum.all?(caps, &is_binary/1) ->
|
|
:ok
|
|
|
|
true ->
|
|
{:error, {:invalid_capabilities, "All capabilities must be strings"}}
|
|
end
|
|
end
|
|
|
|
defp validate_capabilities(_), do: {:error, {:invalid_capabilities, "Capabilities must be a list"}}
|
|
|
|
# Validate OID values map
|
|
defp validate_oid_values(oid_values) when is_map(oid_values) do
|
|
if map_size(oid_values) <= @max_oid_values do
|
|
:ok
|
|
else
|
|
{:error, {:too_many_oids, "OID values map exceeds #{@max_oid_values} entries"}}
|
|
end
|
|
end
|
|
|
|
defp validate_oid_values(_), do: {:error, {:invalid_oid_values, "OID values must be a map"}}
|
|
|
|
# Validate MikroTik sentences
|
|
defp validate_mikrotik_sentences(sentences) when is_list(sentences) do
|
|
if length(sentences) <= @max_mikrotik_sentences do
|
|
:ok
|
|
else
|
|
{:error, {:too_many_sentences, "Sentences list exceeds #{@max_mikrotik_sentences} items"}}
|
|
end
|
|
end
|
|
|
|
defp validate_mikrotik_sentences(_), do: {:error, {:invalid_sentences, "Sentences must be a list"}}
|
|
end
|