defmodule SnmpKit.SnmpLib.Utils do @moduledoc """ Common utilities for SNMP operations including pretty printing, data formatting, timing utilities, and validation functions. This module provides helpful utilities for debugging, logging, monitoring, and general SNMP data manipulation that are commonly needed across SNMP applications. ## Pretty Printing Format SNMP data structures for human-readable display in logs, CLI tools, and debugging output. ## Data Formatting Convert between different representations of SNMP data, format numeric values, and handle common data transformations. ## Timing Utilities Measure and format timing information for SNMP operations, useful for performance monitoring and debugging. ## Validation Common validation functions for SNMP-related data. ## Usage Examples ### Pretty Printing # Format PDU for logging pdu = %{type: :get_request, request_id: 123, varbinds: [...]} Logger.info(SnmpKit.SnmpLib.Utils.pretty_print_pdu(pdu)) # Format individual values value = {:counter32, 12345} IO.puts(SnmpKit.SnmpLib.Utils.pretty_print_value(value)) ### Data Formatting # Format large numbers with separators SnmpKit.SnmpLib.Utils.format_bytes(1048576) # => "1.0 MB" # Format hex strings for MAC addresses SnmpKit.SnmpLib.Utils.format_hex(<<0x00, 0x1B, 0x21, 0x3C, 0x92, 0xEB>>) # => "00:1B:21:3C:92:EB" ### Timing # Time an operation {result, time_us} = SnmpKit.SnmpLib.Utils.measure_request_time(fn -> SnmpKit.SnmpLib.PDU.encode_message(pdu) end) formatted_time = SnmpKit.SnmpLib.Utils.format_response_time(time_us) """ alias SnmpKit.SnmpLib.Error require Logger @formatting :snmpkit@formatting @type oid :: [non_neg_integer()] @type snmp_value :: any() @type varbind :: {oid(), snmp_value()} @type varbinds :: [varbind()] @type pdu :: map() ## Pretty Printing Functions @doc """ Pretty prints an SNMP PDU for human-readable display. Formats the PDU structure with proper indentation and readable field names, suitable for logging, debugging, or CLI display. ## Parameters - `pdu`: SNMP PDU map containing type, request_id, error_status, etc. ## Returns Formatted string representation of the PDU. ## Examples iex> pdu = %{type: :get_request, request_id: 123, varbinds: [{[1,3,6,1,2,1,1,1,0], :null}]} iex> result = SnmpKit.SnmpLib.Utils.pretty_print_pdu(pdu) iex> String.contains?(result, "GET Request") true """ @spec pretty_print_pdu(pdu()) :: String.t() def pretty_print_pdu(pdu) when is_map(pdu) do type_str = format_pdu_type(Map.get(pdu, :type, :unknown)) request_id = Map.get(pdu, :request_id, 0) lines = [ "#{type_str} (ID: #{request_id})" ] lines = add_error_info(lines, pdu) lines = add_bulk_info(lines, pdu) lines = add_varbinds(lines, Map.get(pdu, :varbinds, [])) Enum.join(lines, "\n") end def pretty_print_pdu(_), do: "Invalid PDU" @doc """ Pretty prints a list of varbinds for display. ## Examples iex> varbinds = [{[1,3,6,1,2,1,1,1,0], "Linux server"}] iex> result = SnmpKit.SnmpLib.Utils.pretty_print_varbinds(varbinds) iex> String.contains?(result, "1.3.6.1.2.1.1.1.0") true """ @spec pretty_print_varbinds(varbinds()) :: String.t() def pretty_print_varbinds(varbinds) when is_list(varbinds) do if varbinds == [] do " (no varbinds)" else varbinds |> Enum.with_index(1) |> Enum.map_join("\n", fn {{oid, value}, index} -> " #{index}. #{pretty_print_oid(oid)} = #{pretty_print_value(value)}" end) end end def pretty_print_varbinds(_), do: "Invalid varbinds" @doc """ Pretty prints an OID for display. ## Examples iex> SnmpKit.SnmpLib.Utils.pretty_print_oid([1,3,6,1,2,1,1,1,0]) "1.3.6.1.2.1.1.1.0" iex> SnmpKit.SnmpLib.Utils.pretty_print_oid("1.3.6.1.2.1.1.1.0") "1.3.6.1.2.1.1.1.0" """ @spec pretty_print_oid(oid() | String.t()) :: String.t() def pretty_print_oid(oid) when is_list(oid) do Enum.join(oid, ".") end def pretty_print_oid(oid) when is_binary(oid) do oid end def pretty_print_oid(_), do: "invalid-oid" @doc """ Pretty prints an SNMP value for display. Formats SNMP values with appropriate type information and human-readable representations. ## Examples iex> SnmpKit.SnmpLib.Utils.pretty_print_value({:counter32, 12345}) "Counter32: 12,345" iex> SnmpKit.SnmpLib.Utils.pretty_print_value({:octet_string, "Hello"}) "OCTET STRING: \\"Hello\\"" iex> SnmpKit.SnmpLib.Utils.pretty_print_value(:null) "NULL" """ @spec pretty_print_value(snmp_value()) :: String.t() def pretty_print_value(:null), do: "NULL" def pretty_print_value({:integer, value}), do: "INTEGER: #{value}" def pretty_print_value({:octet_string, value}) when is_binary(value) do if String.printable?(value) do ~s(OCTET STRING: "#{value}") else hex_str = format_hex(value, " ") "OCTET STRING: #{hex_str}" end end def pretty_print_value({:object_identifier, oid}), do: "OID: #{pretty_print_oid(oid)}" def pretty_print_value({:counter32, value}), do: "Counter32: #{format_number(value)}" def pretty_print_value({:gauge32, value}), do: "Gauge32: #{format_number(value)}" def pretty_print_value({:timeticks, value}), do: "TimeTicks: #{format_timeticks(value)}" def pretty_print_value({:counter64, value}), do: "Counter64: #{format_number(value)}" def pretty_print_value({:ip_address, <>}), do: "IpAddress: #{a}.#{b}.#{c}.#{d}" def pretty_print_value({:opaque, data}), do: "Opaque: #{format_hex(data, " ")}" def pretty_print_value({:no_such_object, _}), do: "noSuchObject" def pretty_print_value({:no_such_instance, _}), do: "noSuchInstance" def pretty_print_value({:end_of_mib_view, _}), do: "endOfMibView" def pretty_print_value(value) when is_integer(value), do: "INTEGER: #{value}" def pretty_print_value(value) when is_binary(value) do if String.printable?(value) do ~s("#{value}") else format_hex(value, " ") end end def pretty_print_value(value), do: "#{inspect(value)}" ## Data Formatting Functions @doc """ Formats byte counts in human-readable units. ## Examples iex> SnmpKit.SnmpLib.Utils.format_bytes(1024) "1.0 KB" iex> SnmpKit.SnmpLib.Utils.format_bytes(1048576) "1.0 MB" iex> SnmpKit.SnmpLib.Utils.format_bytes(512) "512 B" """ @spec format_bytes(non_neg_integer()) :: String.t() def format_bytes(bytes) when is_integer(bytes) and bytes >= 0 do @formatting.format_bytes(bytes) end def format_bytes(_), do: "Invalid byte count" @doc """ Formats rates with units. ## Examples iex> SnmpKit.SnmpLib.Utils.format_rate(1500, "bps") "1.5 Kbps" iex> SnmpKit.SnmpLib.Utils.format_rate(45, "pps") "45 pps" """ @spec format_rate(number(), String.t()) :: String.t() def format_rate(value, unit) when is_integer(value) and is_binary(unit) do @formatting.format_rate(value, unit) end def format_rate(value, unit) when is_float(value) and is_binary(unit) do cond do value >= 1_000_000_000 -> "#{Float.round(value / 1_000_000_000, 1)} G#{unit}" value >= 1_000_000 -> "#{Float.round(value / 1_000_000, 1)} M#{unit}" value >= 1_000 -> "#{Float.round(value / 1_000, 1)} K#{unit}" true -> "#{value} #{unit}" end end def format_rate(_, _), do: "Invalid rate" @doc """ Truncates a string to maximum length with ellipsis. ## Examples iex> SnmpKit.SnmpLib.Utils.truncate_string("Hello, World!", 10) "Hello, ..." iex> SnmpKit.SnmpLib.Utils.truncate_string("Short", 10) "Short" """ @spec truncate_string(String.t(), pos_integer()) :: String.t() def truncate_string(string, max_length) when is_binary(string) and is_integer(max_length) and max_length > 3 do @formatting.truncate_string(string, max_length) end def truncate_string(string, _) when is_binary(string), do: string def truncate_string(_, _), do: "" @doc """ Formats binary data as hexadecimal string. ## Parameters - `data`: Binary data to format - `separator`: String to use between hex bytes (default: ":") ## Examples iex> SnmpKit.SnmpLib.Utils.format_hex(<<0x00, 0x1B, 0x21>>) "00:1B:21" iex> SnmpKit.SnmpLib.Utils.format_hex(<<0xDE, 0xAD, 0xBE, 0xEF>>, " ") "DE AD BE EF" """ @spec format_hex(binary(), String.t()) :: String.t() def format_hex(data, separator \\ ":") def format_hex(data, separator) when is_binary(data) and is_binary(separator) do data |> :binary.bin_to_list() |> Enum.map_join(separator, &(&1 |> Integer.to_string(16) |> String.pad_leading(2, "0") |> String.upcase())) end def format_hex(data, _) when not is_binary(data), do: "Invalid binary data" def format_hex(_, separator) when not is_binary(separator), do: "Invalid binary data" @doc """ Formats large numbers with thousand separators. ## Examples iex> SnmpKit.SnmpLib.Utils.format_number(1234567) "1,234,567" iex> SnmpKit.SnmpLib.Utils.format_number(42) "42" """ @spec format_number(integer()) :: String.t() def format_number(number) when is_integer(number) do @formatting.format_number(number) end def format_number(_), do: "Invalid number" ## Timing Utilities @doc """ Measures the execution time of a function in microseconds. ## Parameters - `fun`: Function to execute and time ## Returns Tuple of `{result, time_microseconds}` where result is the function's return value and time_microseconds is the execution time. ## Examples iex> {result, time} = SnmpKit.SnmpLib.Utils.measure_request_time(fn -> :timer.sleep(100); :ok end) iex> result :ok iex> time > 100_000 true """ @spec measure_request_time(function()) :: {any(), non_neg_integer()} def measure_request_time(fun) when is_function(fun) do start_time = System.monotonic_time(:microsecond) result = fun.() end_time = System.monotonic_time(:microsecond) {result, end_time - start_time} end @doc """ Formats response time in human-readable units. ## Parameters - `microseconds`: Time in microseconds ## Examples iex> SnmpKit.SnmpLib.Utils.format_response_time(1500) "1.50ms" iex> SnmpKit.SnmpLib.Utils.format_response_time(2_500_000) "2.50s" iex> SnmpKit.SnmpLib.Utils.format_response_time(500) "500μs" """ @spec format_response_time(non_neg_integer()) :: String.t() def format_response_time(microseconds) when is_integer(microseconds) and microseconds >= 0 do @formatting.format_response_time(microseconds) end def format_response_time(_), do: "Invalid time" ## Target Parsing Functions @doc """ Parses SNMP target specifications into standardized format. Accepts various input formats and returns a consistent target map with host and port. IP addresses are resolved to tuples when possible, hostnames remain as strings. Default port is 161 when not specified. ## Parameters - `target`: Target specification in various formats ## Accepted Input Formats - `"192.168.1.1:161"` - IP with port - `"192.168.1.1"` - IP without port (uses default 161) - `"device.local:162"` - hostname with port - `"device.local"` - hostname without port (uses default 161) - `{192, 168, 1, 1}` - IP tuple (uses default port 161) - `%{host: "192.168.1.1", port: 161}` - already parsed map ## Returns - `{:ok, %{host: host, port: port}}` - Successfully parsed target - `{:error, reason}` - Parse error with reason ## Examples iex> SnmpKit.SnmpLib.Utils.parse_target("192.168.1.1:161") {:ok, %{host: {192, 168, 1, 1}, port: 161}} iex> SnmpKit.SnmpLib.Utils.parse_target("192.168.1.1") {:ok, %{host: {192, 168, 1, 1}, port: 161}} iex> SnmpKit.SnmpLib.Utils.parse_target("device.local:162") {:ok, %{host: "device.local", port: 162}} iex> SnmpKit.SnmpLib.Utils.parse_target("device.local") {:ok, %{host: "device.local", port: 161}} iex> SnmpKit.SnmpLib.Utils.parse_target({192, 168, 1, 1}) {:ok, %{host: {192, 168, 1, 1}, port: 161}} iex> SnmpKit.SnmpLib.Utils.parse_target(%{host: "192.168.1.1", port: 161}) {:ok, %{host: {192, 168, 1, 1}, port: 161}} iex> SnmpKit.SnmpLib.Utils.parse_target("invalid:99999") {:error, {:invalid_port, "99999"}} """ @spec parse_target(String.t() | tuple() | map()) :: {:ok, %{host: :inet.ip_address() | String.t(), port: pos_integer()}} | {:error, any()} def parse_target(target) when is_binary(target) do # Handle IPv6 addresses that might contain colons case parse_host_port_string(target) do {host_str, nil} -> # No port specified, use default parse_host_with_port(host_str, 161) {host_str, port_str} -> # Port specified, validate it case parse_port(port_str) do {:ok, port} -> parse_host_with_port(host_str, port) {:error, reason} -> {:error, reason} end end end def parse_target({a, b, c, d} = ip_tuple) when is_integer(a) and is_integer(b) and is_integer(c) and is_integer(d) do if valid_ip_tuple?(ip_tuple) do {:ok, %{host: ip_tuple, port: 161}} else {:error, {:invalid_ip_tuple, ip_tuple}} end end def parse_target({a, b, c, d, e, f, g, h} = ipv6_tuple) when is_integer(a) and is_integer(b) and is_integer(c) and is_integer(d) and is_integer(e) and is_integer(f) and is_integer(g) and is_integer(h) do if valid_ipv6_tuple?(ipv6_tuple) do {:ok, %{host: ipv6_tuple, port: 161}} else {:error, {:invalid_ipv6_tuple, ipv6_tuple}} end end def parse_target(%{host: host, port: port}) when is_integer(port) do if port > 0 and port <= 65_535 do case parse_host(host) do {:ok, parsed_host} -> {:ok, %{host: parsed_host, port: port}} {:error, reason} -> {:error, reason} end else {:error, {:invalid_port, Integer.to_string(port)}} end end def parse_target(%{host: host}) do case parse_host(host) do {:ok, parsed_host} -> {:ok, %{host: parsed_host, port: 161}} {:error, reason} -> {:error, reason} end end def parse_target(invalid) do {:error, {:invalid_target_format, invalid}} end ## Validation Functions @doc """ Validates an SNMP version number. ## Examples iex> SnmpKit.SnmpLib.Utils.valid_snmp_version?(1) true iex> SnmpKit.SnmpLib.Utils.valid_snmp_version?(5) false """ @spec valid_snmp_version?(any()) :: boolean() def valid_snmp_version?(version) when version in [0, 1, 2, 3], do: true def valid_snmp_version?(:v1), do: true def valid_snmp_version?(:v2c), do: true def valid_snmp_version?(:v3), do: true def valid_snmp_version?(_), do: false @doc """ Validates an SNMP community string. Community strings should be non-empty and contain only printable characters. ## Examples iex> SnmpKit.SnmpLib.Utils.valid_community_string?("public") true iex> SnmpKit.SnmpLib.Utils.valid_community_string?("") false """ @spec valid_community_string?(any()) :: boolean() def valid_community_string?(community) when is_binary(community) do byte_size(community) > 0 and String.printable?(community) end def valid_community_string?(_), do: false @doc """ Sanitizes a community string for safe logging. Replaces community strings with asterisks to prevent credential leakage in logs while preserving length information. ## Examples iex> SnmpKit.SnmpLib.Utils.sanitize_community("secret123") "*********" iex> SnmpKit.SnmpLib.Utils.sanitize_community("") "" """ @spec sanitize_community(String.t()) :: String.t() def sanitize_community(community) when is_binary(community) do case byte_size(community) do 0 -> "" size -> String.duplicate("*", size) end end def sanitize_community(_), do: "" ## Private Helper Functions # Target parsing helpers defp parse_host_port_string(target_str) do cond do # RFC 3986 bracket notation: [IPv6]:port String.starts_with?(target_str, "[") -> parse_bracket_notation(target_str) # Check if this looks like plain IPv6 (contains :: or multiple colons) String.contains?(target_str, "::") or count_colons(target_str) > 1 -> # Plain IPv6 address without port {target_str, nil} # Standard host:port parsing for IPv4 and simple hostnames true -> parse_standard_host_port(target_str) end end defp parse_standard_host_port(target_str) do case String.split(target_str, ":", parts: 2) do [host_str] -> {host_str, nil} [host_str, port_str] -> # Check if host part looks like IPv4 or simple hostname if ipv4_or_simple_hostname?(host_str) do {host_str, port_str} else # Complex hostname with colons, treat as hostname without port {target_str, nil} end end end defp parse_bracket_notation("[" <> rest) do case String.split(rest, "]:", parts: 2) do [ipv6_addr, port_str] -> # Valid [IPv6]:port format {ipv6_addr, port_str} _ -> # Check for [IPv6] without port case String.split(rest, "]", parts: 2) do [ipv6_addr, ""] -> # Valid [IPv6] format without port {ipv6_addr, nil} _ -> # Invalid bracket notation, treat as hostname {"[" <> rest, nil} end end end defp count_colons(str) do str |> String.graphemes() |> Enum.count(&(&1 == ":")) end defp parse_host_with_port(host_str, port) do case parse_host(host_str) do {:ok, host} -> {:ok, %{host: host, port: port}} {:error, reason} -> {:error, reason} end end defp parse_host(host) when is_binary(host) do case :inet.parse_address(String.to_charlist(host)) do {:ok, {a, b, c, d}} when is_integer(a) and is_integer(b) and is_integer(c) and is_integer(d) -> # IPv4 address {:ok, {a, b, c, d}} {:ok, ipv6_tuple} when tuple_size(ipv6_tuple) == 8 -> # IPv6 address - return as tuple for proper socket handling {:ok, ipv6_tuple} {:error, :einval} -> # Not an IP, treat as hostname {:ok, host} end end defp parse_host({a, b, c, d} = ip_tuple) when is_integer(a) and is_integer(b) and is_integer(c) and is_integer(d) do if valid_ip_tuple?(ip_tuple) do {:ok, ip_tuple} else {:error, {:invalid_ip_tuple, ip_tuple}} end end defp parse_host(invalid) do {:error, {:invalid_host_format, invalid}} end defp parse_port(port_str) when is_binary(port_str) do case Integer.parse(port_str) do {port, ""} when port > 0 and port <= 65_535 -> {:ok, port} {_port, ""} -> {:error, {:invalid_port, port_str}} _ -> {:error, {:invalid_port_format, port_str}} end end defp valid_ip_tuple?({a, b, c, d}) do Enum.all?([a, b, c, d], &valid_ipv4_octet?/1) end defp valid_ip_tuple?(_), do: false defp valid_ipv4_octet?(n), do: is_integer(n) and n >= 0 and n <= 255 defp valid_ipv6_tuple?({a, b, c, d, e, f, g, h}) do Enum.all?([a, b, c, d, e, f, g, h], &valid_ipv6_segment?/1) end defp valid_ipv6_tuple?(_), do: false defp valid_ipv6_segment?(n), do: is_integer(n) and n >= 0 and n <= 65_535 defp ipv4_or_simple_hostname?(host_str) do # Check if it looks like IPv4 (has 3 dots and only digits/dots) case String.split(host_str, ".") do [a, b, c, d] -> Enum.all?([a, b, c, d], &valid_ipv4_octet_string?/1) _ -> simple_hostname?(host_str) end end defp valid_ipv4_octet_string?(part) do case Integer.parse(part) do {num, ""} when num >= 0 and num <= 255 -> true _ -> false end end defp simple_hostname?(host_str) do # Not IPv4 format, check if it's a simple hostname # Exclude anything that looks like IPv6 (contains :: or multiple colons) not String.contains?(host_str, "::") and count_colons(host_str) <= 1 and not String.contains?(host_str, ":") and String.match?(host_str, ~r/^[a-zA-Z0-9\-\.\_]+$/) end defp format_pdu_type(:get_request), do: "GET Request" defp format_pdu_type(:get_next_request), do: "GET-NEXT Request" defp format_pdu_type(:get_bulk_request), do: "GET-BULK Request" defp format_pdu_type(:set_request), do: "SET Request" defp format_pdu_type(:get_response), do: "Response" defp format_pdu_type(:trap), do: "Trap" defp format_pdu_type(:inform_request), do: "Inform Request" defp format_pdu_type(:snmpv2_trap), do: "SNMPv2 Trap" defp format_pdu_type(:report), do: "Report" defp format_pdu_type(type), do: "#{type}" defp add_error_info(lines, pdu) do error_status = Map.get(pdu, :error_status, 0) error_index = Map.get(pdu, :error_index, 0) if error_status == 0 do lines else error_name = if function_exported?(Error, :error_name, 1) do Error.error_name(error_status) else "error_#{error_status}" end lines ++ [" Error: #{error_name} (#{error_status}) at index #{error_index}"] end end defp add_bulk_info(lines, pdu) do case Map.get(pdu, :type) do :get_bulk_request -> non_repeaters = Map.get(pdu, :non_repeaters, 0) max_repetitions = Map.get(pdu, :max_repetitions, 0) lines ++ [" Non-repeaters: #{non_repeaters}, Max-repetitions: #{max_repetitions}"] _ -> lines end end defp add_varbinds(lines, varbinds) do varbind_str = pretty_print_varbinds(varbinds) lines ++ ["Varbinds:", varbind_str] end defp format_timeticks(ticks) when is_integer(ticks) do # Convert centiseconds to readable time format total_seconds = div(ticks, 100) days = div(total_seconds, 86_400) hours = div(rem(total_seconds, 86_400), 3600) minutes = div(rem(total_seconds, 3600), 60) seconds = rem(total_seconds, 60) time_parts = [] time_parts = if days > 0, do: time_parts ++ ["#{days}d"], else: time_parts time_parts = if hours > 0, do: time_parts ++ ["#{hours}h"], else: time_parts time_parts = if minutes > 0, do: time_parts ++ ["#{minutes}m"], else: time_parts time_parts = if seconds > 0 or time_parts == [], do: time_parts ++ ["#{seconds}s"], else: time_parts "#{format_number(ticks)} (#{Enum.join(time_parts, " ")})" end defp format_timeticks(_), do: "Invalid timeticks" end