graham/towerops
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

credo fixes

Browse source
This commit is contained in:
Graham McIntire 2026-01-23 16:49:02 -06:00
parent 2b78b1a2d3
commit 29f00c37ed
No known key found for this signature in database
12 changed files with 507 additions and 500 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
lib/snmpkit/snmp_lib/asn1.ex
View file

@ -470,29 +470,32 @@ defmodule SnmpKit.SnmpLib.ASN1 do
end
def decode_length(<<length_byte, rest::binary>>) when length_byte > @short_form_max do
if length_byte == @indefinite_length do
{:error, :indefinite_length_not_supported}
else
# Long form
num_octets = length_byte - @indefinite_length
cond do
length_byte == @indefinite_length ->
{:error, :indefinite_length_not_supported}
if num_octets > 4 do
length_byte - @indefinite_length > 4 ->
{:error, :length_too_large}
else
case rest do
<<length_bytes::binary-size(num_octets), remaining::binary>> ->
length = :binary.decode_unsigned(length_bytes, :big)
{:ok, {length, remaining}}
_ ->
{:error, :insufficient_length_bytes}
end
end
true ->
num_octets = length_byte - @indefinite_length
decode_long_form_length(rest, num_octets)
end
end
def decode_length(_), do: {:error, :insufficient_data}
defp decode_long_form_length(data, num_octets) do
case data do
<<length_bytes::binary-size(num_octets), remaining::binary>> ->
length = :binary.decode_unsigned(length_bytes, :big)
{:ok, {length, remaining}}
_ ->
{:error, :insufficient_length_bytes}
end
end
## Tag Parsing
@doc """

30
lib/snmpkit/snmp_lib/host_parser.ex
View file

@ -96,6 +96,7 @@ defmodule SnmpKit.SnmpLib.HostParser do
end
# IPv6 tuple without port
# credo:disable-for-next-line Credo.Check.Refactor.CyclomaticComplexity
def parse({a, b, c, d, e, f, g, h} = ip_tuple, default_port)
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
@ -260,27 +261,16 @@ defmodule SnmpKit.SnmpLib.HostParser do
end
defp parse_ip_without_port(input, default_port) do
cond do
# Try IPv4 first
String.contains?(input, ".") ->
case parse_ipv4_address(input) do
{:ok, ip_tuple} -> {:ok, {ip_tuple, default_port}}
error -> error
end
parser =
cond do
String.contains?(input, ".") -> &parse_ipv4_address/1
String.contains?(input, ":") -> &parse_ipv6_address/1
true -> &resolve_hostname/1
end
# Try IPv6
String.contains?(input, ":") ->
case parse_ipv6_address(input) do
{:ok, ip_tuple} -> {:ok, {ip_tuple, default_port}}
error -> error
end
# Try hostname resolution
true ->
case resolve_hostname(input) do
{:ok, ip_tuple} -> {:ok, {ip_tuple, default_port}}
error -> error
end
case parser.(input) do
{:ok, ip_tuple} -> {:ok, {ip_tuple, default_port}}
error -> error
end
end

2
lib/snmpkit/snmp_lib/mib/parser.ex
View file

@ -39,6 +39,7 @@ defmodule SnmpKit.SnmpLib.MIB.Parser do
# Compile the grammar using Erlang's yecc (if available)
if Code.ensure_loaded?(:yecc) and function_exported?(:yecc, :file, 1) do
# credo:disable-for-next-line Credo.Check.Refactor.Apply
result = apply(:yecc, :file, [to_charlist(grammar_file)])
case result do
@ -141,6 +142,7 @@ defmodule SnmpKit.SnmpLib.MIB.Parser do
end
defp parse_with_module(parser_module, tokens) do
# credo:disable-for-next-line Credo.Check.Refactor.Apply
case apply(parser_module, :parse, [tokens]) do
{:ok, parse_tree} -> {:ok, parse_tree}
{:error, reason} -> {:error, {:parse, reason}}

63
lib/snmpkit/snmp_lib/mib/registry.ex
View file

@ -499,42 +499,41 @@ defmodule SnmpKit.SnmpLib.MIB.Registry do
end
defp find_children(parent_oid, name_to_oid_map) do
normalized_oid =
cond do
is_nil(parent_oid) ->
[]
case normalize_parent_oid(parent_oid) do
{:ok, normalized_oid} ->
children = filter_direct_children(name_to_oid_map, normalized_oid)
{:ok, children}
is_binary(parent_oid) ->
case OID.string_to_list(parent_oid) do
{:ok, oid_list} -> oid_list
{:error, _} -> []
end
is_list(parent_oid) ->
parent_oid
true ->
[]
end
# Return error for invalid OIDs
if normalized_oid == [] and not is_nil(parent_oid) do
{:error, :invalid_parent_oid}
else
children =
name_to_oid_map
|> Enum.filter(fn {_name, oid} ->
is_list(oid) and is_list(normalized_oid) and
length(oid) == length(normalized_oid) + 1 and
List.starts_with?(oid, normalized_oid)
end)
|> Enum.map(fn {name, _oid} -> name end)
|> Enum.sort()
{:ok, children}
{:error, reason} ->
{:error, reason}
end
end
defp normalize_parent_oid(nil), do: {:ok, []}
defp normalize_parent_oid(oid) when is_list(oid), do: {:ok, oid}
defp normalize_parent_oid(oid) when is_binary(oid) do
case OID.string_to_list(oid) do
{:ok, oid_list} -> {:ok, oid_list}
{:error, _} -> {:error, :invalid_parent_oid}
end
end
defp normalize_parent_oid(_), do: {:error, :invalid_parent_oid}
defp filter_direct_children(name_to_oid_map, parent_oid) do
name_to_oid_map
|> Enum.filter(&direct_child?(&1, parent_oid))
|> Enum.map(fn {name, _oid} -> name end)
|> Enum.sort()
end
defp direct_child?({_name, oid}, parent_oid) when is_list(oid) and is_list(parent_oid) do
length(oid) == length(parent_oid) + 1 and List.starts_with?(oid, parent_oid)
end
defp direct_child?(_, _), do: false
defp walk_tree_from_root(root_oid, name_to_oid_map) do
root_oid =
cond do

15
lib/snmpkit/snmp_lib/mib/utilities.ex
View file

@ -104,19 +104,8 @@ defmodule SnmpKit.SnmpLib.MIB.Utilities do
Port of allow_size_rfc1902/1 from snmpc_lib.erl
"""
@spec allow_size_rfc1902(atom()) :: boolean()
def allow_size_rfc1902(type) do
case type do
:octet_string -> true
:object_identifier -> false
:integer -> false
:counter32 -> false
:counter64 -> false
:gauge32 -> false
:time_ticks -> false
:ip_address -> false
_ -> false
end
end
def allow_size_rfc1902(:octet_string), do: true
def allow_size_rfc1902(_), do: false
@doc """
Validate sub-identifier ranges.

65
lib/snmpkit/snmp_lib/monitor.ex
View file

@ -746,39 +746,44 @@ defmodule SnmpKit.SnmpLib.Monitor do
# Check if alert already exists
alert_key = {threshold.device_id, threshold.metric}
case Enum.find(existing_alerts, fn alert ->
{alert.device_id, alert.metric} == alert_key
end) do
nil ->
# New alert
new_alert = %{
device_id: threshold.device_id,
metric: threshold.metric,
threshold: threshold.threshold,
current_value: get_current_metric_value(metric, threshold.metric),
fired_at: System.monotonic_time(:millisecond),
callback: threshold.callback
}
# Execute callback if provided
if threshold.callback do
spawn(fn -> threshold.callback.(new_alert) end)
end
Logger.warning(
"Alert fired: " <>
threshold.device_id <>
" " <> to_string(threshold.metric) <> " = " <> to_string(new_alert.current_value)
)
[new_alert | existing_alerts]
_existing ->
# Alert already active
existing_alerts
if alert_exists?(existing_alerts, alert_key) do
existing_alerts
else
create_and_fire_new_alert(threshold, metric, existing_alerts)
end
end
defp alert_exists?(alerts, alert_key) do
Enum.any?(alerts, fn alert -> {alert.device_id, alert.metric} == alert_key end)
end
defp create_and_fire_new_alert(threshold, metric, existing_alerts) do
new_alert = %{
device_id: threshold.device_id,
metric: threshold.metric,
threshold: threshold.threshold,
current_value: get_current_metric_value(metric, threshold.metric),
fired_at: System.monotonic_time(:millisecond),
callback: threshold.callback
}
execute_alert_callback(threshold.callback, new_alert)
log_alert_fired(new_alert)
[new_alert | existing_alerts]
end
defp execute_alert_callback(nil, _alert), do: :ok
defp execute_alert_callback(callback, alert) do
spawn(fn -> callback.(alert) end)
:ok
end
defp log_alert_fired(alert) do
Logger.warning("Alert fired: #{alert.device_id} #{alert.metric} = #{alert.current_value}")
end
defp get_current_metric_value(metric, :response_time), do: metric.duration
defp get_current_metric_value(_metric, _metric_type), do: nil

53
lib/snmpkit/snmp_lib/oid.ex
View file

@ -102,43 +102,32 @@ defmodule SnmpKit.SnmpLib.OID do
"""
@spec string_to_list(oid_string()) :: {:ok, oid()} | {:error, atom()}
def string_to_list(oid_string) when is_binary(oid_string) do
case String.trim(oid_string) do
"" ->
{:error, :empty_oid}
trimmed_string ->
# Handle leading dot by removing it
normalized_string =
if String.starts_with?(trimmed_string, ".") do
String.slice(trimmed_string, 1..-1//1)
else
trimmed_string
end
# Check if we're left with an empty string after removing leading dot
case normalized_string do
"" ->
{:error, :empty_oid}
_ ->
parts = String.split(normalized_string, ".")
case parse_oid_components(parts) do
{:ok, oid_list} ->
case validate_oid_list(oid_list) do
:ok -> {:ok, oid_list}
{:error, reason} -> {:error, reason}
end
{:error, reason} ->
{:error, reason}
end
end
with {:ok, normalized} <- normalize_oid_string(oid_string),
parts = String.split(normalized, "."),
{:ok, oid_list} <- parse_oid_components(parts),
:ok <- validate_oid_list(oid_list) do
{:ok, oid_list}
end
end
def string_to_list(_), do: {:error, :invalid_input}
defp normalize_oid_string(oid_string) do
trimmed = String.trim(oid_string)
cond do
trimmed == "" ->
{:error, :empty_oid}
String.starts_with?(trimmed, ".") ->
normalized = String.slice(trimmed, 1..-1//1)
if normalized == "", do: {:error, :empty_oid}, else: {:ok, normalized}
true ->
{:ok, trimmed}
end
end
@doc """
Converts an OID list to a dot-separated string.

105
lib/snmpkit/snmp_lib/pdu/builder.ex
View file

@ -195,51 +195,53 @@ defmodule SnmpKit.SnmpLib.PDU.Builder do
"""
@spec validate(pdu()) :: {:ok, pdu()} | {:error, atom()}
def validate(pdu) when is_map(pdu) do
# First check if we have a type field
case Map.get(pdu, :type) do
nil ->
{:error, :missing_required_fields}
type ->
# Validate the type first
case validate_pdu_type_only(type) do
:ok ->
# Now check required fields based on type
basic_fields = [:request_id, :varbinds]
if Enum.all?(basic_fields, &Map.has_key?(pdu, &1)) do
case type do
:get_bulk_request ->
bulk_fields = [:non_repeaters, :max_repetitions]
if Enum.all?(bulk_fields, &Map.has_key?(pdu, &1)) do
{:ok, pdu}
else
{:error, :missing_bulk_fields}
end
_ ->
# Standard PDUs need error_status and error_index
standard_fields = [:error_status, :error_index]
if Enum.all?(standard_fields, &Map.has_key?(pdu, &1)) do
{:ok, pdu}
else
{:error, :missing_required_fields}
end
end
else
{:error, :missing_required_fields}
end
:error ->
{:error, :invalid_pdu_type}
end
with {:ok, type} <- get_pdu_type(pdu),
:ok <- validate_pdu_type_only(type),
:ok <- validate_basic_fields(pdu),
:ok <- validate_type_specific_fields(pdu, type) do
{:ok, pdu}
end
end
def validate(_), do: {:error, :invalid_pdu_format}
defp get_pdu_type(pdu) do
case Map.get(pdu, :type) do
nil -> {:error, :missing_required_fields}
type -> {:ok, type}
end
end
defp validate_basic_fields(pdu) do
basic_fields = [:request_id, :varbinds]
if Enum.all?(basic_fields, &Map.has_key?(pdu, &1)) do
:ok
else
{:error, :missing_required_fields}
end
end
defp validate_type_specific_fields(pdu, :get_bulk_request) do
bulk_fields = [:non_repeaters, :max_repetitions]
if Enum.all?(bulk_fields, &Map.has_key?(pdu, &1)) do
:ok
else
{:error, :missing_bulk_fields}
end
end
defp validate_type_specific_fields(pdu, _type) do
standard_fields = [:error_status, :error_index]
if Enum.all?(standard_fields, &Map.has_key?(pdu, &1)) do
:ok
else
{:error, :missing_required_fields}
end
end
@doc """
Validates a community string against an encoded SNMP message.
"""
@ -301,16 +303,17 @@ defmodule SnmpKit.SnmpLib.PDU.Builder do
# Helper function to validate PDU type
defp validate_pdu_type_only(type) do
case type do
:get_request -> :ok
:get_next_request -> :ok
:get_response -> :ok
:set_request -> :ok
:get_bulk_request -> :ok
:inform_request -> :ok
:snmpv2_trap -> :ok
:report -> :ok
_ -> :error
end
valid_types = [
:get_request,
:get_next_request,
:get_response,
:set_request,
:get_bulk_request,
:inform_request,
:snmpv2_trap,
:report
]
if type in valid_types, do: :ok, else: :error
end
end

92
lib/snmpkit/snmp_lib/pdu/v3_encoder.ex
View file

@ -175,32 +175,9 @@ defmodule SnmpKit.SnmpLib.PDU.V3Encoder do
@spec decode_message(binary(), security_user() | nil) ::
{:ok, v3_message()} | {:error, atom()}
def decode_message(data, user \\ nil) when is_binary(data) do
case decode_v3_message(data) do
{:ok, message, security_params, msg_data} ->
# Apply security processing
case process_security_parameters(message, security_params, msg_data, user) do
{:ok, result} when user == nil ->
# For discovery messages, process_security_parameters already returns the scoped_pdu
final_message = Map.put(message, :msg_data, result)
{:ok, final_message}
{:ok, decrypted_data} ->
# Decode scoped PDU
case decode_scoped_pdu(decrypted_data) do
{:ok, scoped_pdu} ->
final_message = Map.put(message, :msg_data, scoped_pdu)
{:ok, final_message}
{:error, reason} ->
{:error, reason}
end
{:error, reason} ->
{:error, reason}
end
{:error, reason} ->
{:error, reason}
with {:ok, message, security_params, msg_data} <- decode_v3_message(data),
{:ok, result} <- process_security_parameters(message, security_params, msg_data, user) do
finalize_decoded_message(message, result, user)
end
rescue
error ->
@ -208,6 +185,24 @@ defmodule SnmpKit.SnmpLib.PDU.V3Encoder do
{:error, :decoding_failed}
end
defp finalize_decoded_message(message, result, nil) do
# For discovery messages, process_security_parameters already returns the scoped_pdu
final_message = Map.put(message, :msg_data, result)
{:ok, final_message}
end
defp finalize_decoded_message(message, decrypted_data, _user) do
# Decode scoped PDU
case decode_scoped_pdu(decrypted_data) do
{:ok, scoped_pdu} ->
final_message = Map.put(message, :msg_data, scoped_pdu)
{:ok, final_message}
{:error, reason} ->
{:error, reason}
end
end
# Private encoding functions
defp encode_v3_message(message, security_params, msg_data) do
@ -411,38 +406,27 @@ defmodule SnmpKit.SnmpLib.PDU.V3Encoder do
# Private decoding functions
defp decode_v3_message(data) do
case ASN1.decode_sequence(data) do
{:ok, {content, _remaining}} ->
case decode_message_components(content) do
{:ok, version, header_data, security_params, msg_data} when version == 3 ->
case decode_header_data(header_data) do
{:ok, msg_id, msg_max_size, msg_flags, security_model} ->
message = %{
version: version,
msg_id: msg_id,
msg_max_size: msg_max_size,
msg_flags: msg_flags,
msg_security_model: security_model
}
with {:ok, {content, _remaining}} <- ASN1.decode_sequence(data),
{:ok, version, header_data, security_params, msg_data} <-
decode_message_components(content),
:ok <- validate_version(version),
{:ok, msg_id, msg_max_size, msg_flags, security_model} <-
decode_header_data(header_data) do
message = %{
version: version,
msg_id: msg_id,
msg_max_size: msg_max_size,
msg_flags: msg_flags,
msg_security_model: security_model
}
{:ok, message, security_params, msg_data}
{:error, reason} ->
{:error, reason}
end
{:ok, version, _, _, _} ->
{:error, {:invalid_version, version}}
{:error, reason} ->
{:error, reason}
end
{:error, reason} ->
{:error, reason}
{:ok, message, security_params, msg_data}
end
end
defp validate_version(3), do: :ok
defp validate_version(version), do: {:error, {:invalid_version, version}}
defp decode_message_components(data) do
with {:ok, {version, rest1}} <- ASN1.decode_integer(data),
{:ok, {header_data, rest2}} <- ASN1.decode_sequence(rest1),

26
lib/snmpkit/snmp_lib/pool.ex
View file

@ -658,24 +658,22 @@ defmodule SnmpKit.SnmpLib.Pool do
# Create replacement connections
needed = length(unhealthy_connections)
new_connections =
if needed > 0 do
1..needed
|> Enum.map(fn _i ->
case create_connection(state.worker_opts) do
{:ok, conn} -> conn
{:error, _} -> nil
end
end)
|> Enum.filter(& &1)
else
[]
end
new_connections = create_replacement_connections(needed, state.worker_opts)
%{state | connections: healthy_connections ++ new_connections}
end
defp create_replacement_connections(0, _worker_opts), do: []
defp create_replacement_connections(needed, worker_opts) do
1..needed
|> Enum.map(fn _i -> create_connection(worker_opts) end)
|> Enum.flat_map(fn
{:ok, conn} -> [conn]
{:error, _} -> []
end)
end
defp split_connections(connections, main_size) do
{main, overflow} = Enum.split(connections, main_size)
{main, overflow}

10
lib/snmpkit/snmp_lib/types.ex
View file

@ -400,10 +400,12 @@ defmodule SnmpKit.SnmpLib.Types do
:ok
end
def validate_ip_address({a, b, c, d})
when is_integer(a) and is_integer(b) and is_integer(c) and is_integer(d) and a >= 0 and a <= 255 and b >= 0 and
b <= 255 and c >= 0 and c <= 255 and d >= 0 and d <= 255 do
:ok
def validate_ip_address({a, b, c, d}) when is_integer(a) and is_integer(b) and is_integer(c) and is_integer(d) do
if valid_ip_octet?(a) and valid_ip_octet?(b) and valid_ip_octet?(c) and valid_ip_octet?(d) do
:ok
else
{:error, :out_of_range}
end
end
def validate_ip_address(value) when is_binary(value) do

513
lib/snmpkit/snmp_mgr/mib.ex
View file

@ -663,57 +663,52 @@ defmodule SnmpKit.SnmpMgr.MIB do
defp extract_name_to_oid_from_symbols(symbols) when is_map(symbols) do
Enum.reduce(symbols, %{}, fn {name, defn}, acc ->
case defn do
%{} ->
case Map.get(defn, :oid) do
nil ->
acc
oid_any ->
case normalize_parsed_oid(oid_any) do
{:ok, oid_list} -> Map.put(acc, name, oid_list)
_ -> acc
end
end
_ ->
acc
end
extract_single_name_to_oid(name, defn, acc)
end)
end
defp extract_single_name_to_oid(name, %{} = defn, acc) do
case Map.get(defn, :oid) do
nil -> acc
oid_any -> maybe_add_normalized_oid(name, oid_any, acc)
end
end
defp extract_single_name_to_oid(_name, _defn, acc), do: acc
defp maybe_add_normalized_oid(name, oid_any, acc) do
case normalize_parsed_oid(oid_any) do
{:ok, oid_list} -> Map.put(acc, name, oid_list)
_ -> acc
end
end
defp extract_meta_from_symbols(symbols) when is_map(symbols) do
Enum.reduce(symbols, %{}, fn {name, defn}, acc ->
case defn do
%{} ->
case Map.get(defn, :__type__) do
:object_type ->
syntax_any = Map.get(defn, :syntax)
access = Map.get(defn, :max_access)
status = Map.get(defn, :status)
description = Map.get(defn, :description)
meta = %{
syntax_base: syntax_base_from(syntax_any),
textual_convention: textual_convention_from(syntax_any),
display_hint: nil,
access: access,
status: status,
description: description
}
Map.put(acc, name, meta)
_ ->
acc
end
_ ->
acc
end
extract_single_meta(name, defn, acc)
end)
end
defp extract_single_meta(name, %{__type__: :object_type} = defn, acc) do
meta = build_meta_from_object_type(defn)
Map.put(acc, name, meta)
end
defp extract_single_meta(_name, _defn, acc), do: acc
defp build_meta_from_object_type(defn) do
syntax_any = Map.get(defn, :syntax)
%{
syntax_base: syntax_base_from(syntax_any),
textual_convention: textual_convention_from(syntax_any),
display_hint: nil,
access: Map.get(defn, :max_access),
status: Map.get(defn, :status),
description: Map.get(defn, :description)
}
end
defp compile_dir_fallback(directory, opts) do
case File.ls(directory) do
{:ok, files} ->
@ -861,38 +856,52 @@ defmodule SnmpKit.SnmpMgr.MIB do
Map.put(state, :integrated_mibs, new_integrated)
end
defp resolve_name(name, _name_to_oid_map) when is_nil(name) or not is_binary(name) do
{:error, :invalid_name}
end
defp resolve_name(name, name_to_oid_map) do
cond do
# Handle nil or invalid names first
is_nil(name) or not is_binary(name) ->
{:error, :invalid_name}
case Map.get(name_to_oid_map, name) do
nil -> resolve_name_with_instance(name, name_to_oid_map)
oid -> {:ok, oid}
end
end
# Direct match
Map.has_key?(name_to_oid_map, name) ->
{:ok, Map.get(name_to_oid_map, name)}
defp resolve_name_with_instance(name, name_to_oid_map) do
if String.contains?(name, ".") do
parse_name_with_instance(name, name_to_oid_map)
else
{:error, :not_found}
end
end
# Name with instance (e.g., "sysDescr.0")
String.contains?(name, ".") ->
[base_name | instance_parts] = String.split(name, ".")
defp parse_name_with_instance(name, name_to_oid_map) do
[base_name | instance_parts] = String.split(name, ".")
case Map.get(name_to_oid_map, base_name) do
nil ->
{:error, :not_found}
with {:ok, base_oid} <- get_base_oid(base_name, name_to_oid_map),
{:ok, instance_oids} <- parse_instance_parts(instance_parts) do
{:ok, base_oid ++ instance_oids}
end
end
base_oid ->
case Enum.reduce_while(instance_parts, [], fn part, acc ->
case Integer.parse(part) do
{int, ""} -> {:cont, [int | acc]}
_ -> {:halt, :error}
end
end) do
:error -> {:error, :invalid_instance}
instance_oids -> {:ok, base_oid ++ Enum.reverse(instance_oids)}
end
end
defp get_base_oid(base_name, name_to_oid_map) do
case Map.get(name_to_oid_map, base_name) do
nil -> {:error, :not_found}
oid -> {:ok, oid}
end
end
true ->
{:error, :not_found}
defp parse_instance_parts(parts) do
parts
|> Enum.reduce_while([], fn part, acc ->
case Integer.parse(part) do
{int, ""} -> {:cont, [int | acc]}
_ -> {:halt, :error}
end
end)
|> case do
:error -> {:error, :invalid_instance}
oids -> {:ok, Enum.reverse(oids)}
end
end
@ -945,42 +954,46 @@ defmodule SnmpKit.SnmpMgr.MIB do
end
defp find_children(parent_oid, name_to_oid_map) do
normalized_oid =
cond do
is_nil(parent_oid) ->
[]
case normalize_parent_oid_for_children(parent_oid) do
{:ok, normalized_oid} ->
children = filter_and_collect_children(normalized_oid, name_to_oid_map)
{:ok, children}
is_binary(parent_oid) ->
case OID.string_to_list(parent_oid) do
{:ok, oid_list} -> oid_list
{:error, _} -> []
end
is_list(parent_oid) ->
parent_oid
true ->
[]
end
# Return error for invalid OIDs
if normalized_oid == [] and not is_nil(parent_oid) do
{:error, :invalid_parent_oid}
else
children =
name_to_oid_map
|> Enum.filter(fn {_name, oid} ->
is_list(oid) and is_list(normalized_oid) and
length(oid) == length(normalized_oid) + 1 and
List.starts_with?(oid, normalized_oid)
end)
|> Enum.map(fn {name, _oid} -> name end)
|> Enum.sort()
{:ok, children}
{:error, reason} ->
{:error, reason}
end
end
# Normalize parent OID to list format
defp normalize_parent_oid_for_children(nil), do: {:ok, []}
defp normalize_parent_oid_for_children(parent_oid) when is_list(parent_oid), do: {:ok, parent_oid}
defp normalize_parent_oid_for_children(parent_oid) when is_binary(parent_oid) do
case OID.string_to_list(parent_oid) do
{:ok, oid_list} -> {:ok, oid_list}
{:error, _} -> {:error, :invalid_parent_oid}
end
end
defp normalize_parent_oid_for_children(_), do: {:error, :invalid_parent_oid}
# Filter and collect direct children of parent OID
defp filter_and_collect_children(normalized_oid, name_to_oid_map) do
name_to_oid_map
|> Enum.filter(fn {_name, oid} ->
direct_child_of?(oid, normalized_oid)
end)
|> Enum.map(fn {name, _oid} -> name end)
|> Enum.sort()
end
# Check if oid is a direct child of parent_oid
defp direct_child_of?(oid, parent_oid) when is_list(oid) and is_list(parent_oid) do
length(oid) == length(parent_oid) + 1 and List.starts_with?(oid, parent_oid)
end
defp direct_child_of?(_oid, _parent_oid), do: false
defp walk_tree_from_root(root_oid, name_to_oid_map) do
root_oid =
cond do
@ -1033,30 +1046,33 @@ defmodule SnmpKit.SnmpMgr.MIB do
end
defp normalize_to_oid_list(oid_any) when is_binary(oid_any) do
if String.contains?(oid_any, ".") and String.match?(oid_any, ~r/^\.?\d+(?:\.\d+)*$/) do
if numeric_oid_string?(oid_any) do
OID.string_to_list(oid_any)
else
case String.split(oid_any, ".", parts: 2) do
[base] ->
case resolve(base) do
{:ok, base_oid} -> {:ok, base_oid}
error -> error
end
[base, instance_str] ->
with {:ok, base_oid} <- resolve(base),
{:ok, instance_index} <- parse_instance(instance_str) do
{:ok, base_oid ++ instance_index}
else
{:error, _} = err -> err
_ -> {:error, :invalid_instance}
end
end
resolve_named_oid(oid_any)
end
end
defp normalize_to_oid_list(_), do: {:error, :invalid_input}
defp numeric_oid_string?(str) do
String.contains?(str, ".") and String.match?(str, ~r/^\.?\d+(?:\.\d+)*$/)
end
defp resolve_named_oid(oid_str) do
case String.split(oid_str, ".", parts: 2) do
[base] -> resolve(base)
[base, instance_str] -> resolve_with_instance(base, instance_str)
end
end
defp resolve_with_instance(base, instance_str) do
with {:ok, base_oid} <- resolve(base),
{:ok, instance_index} <- parse_instance(instance_str) do
{:ok, base_oid ++ instance_index}
end
end
defp parse_instance(instance_str) do
parts = String.split(instance_str, ".")
@ -1076,27 +1092,20 @@ defmodule SnmpKit.SnmpMgr.MIB do
end
defp base_name_and_index(oid_list) do
case reverse_lookup(oid_list) do
{:ok, name_with_index} ->
# Strip instance suffix to get the true base name (e.g., "sysDescr.0" -> "sysDescr")
base_name = strip_instance_suffix(name_with_index)
with {:ok, name_with_index} <- reverse_lookup(oid_list),
base_name = strip_instance_suffix(name_with_index),
{:ok, base_oid} <- name_to_oid(base_name) do
extract_index_from_oid(oid_list, base_name, base_oid)
end
end
case name_to_oid(base_name) do
{:ok, base_oid} ->
base_len = length(base_oid)
defp extract_index_from_oid(oid_list, base_name, base_oid) do
base_len = length(base_oid)
if length(oid_list) > base_len do
{:ok, base_name, Enum.drop(oid_list, base_len)}
else
{:ok, base_name, nil}
end
{:error, _} = err ->
err
end
{:error, reason} ->
{:error, reason}
if length(oid_list) > base_len do
{:ok, base_name, Enum.drop(oid_list, base_len)}
else
{:ok, base_name, nil}
end
end
@ -1135,18 +1144,23 @@ defmodule SnmpKit.SnmpMgr.MIB do
defp maybe_put(map, key, val), do: Map.put(map, key, val)
defp module_for(base_name) do
cond do
String.starts_with?(base_name, "sys") -> "SNMPv2-MIB"
String.starts_with?(base_name, "ifHC") -> "IF-MIB"
String.starts_with?(base_name, "ifIn") -> "IF-MIB"
String.starts_with?(base_name, "ifOut") -> "IF-MIB"
String.starts_with?(base_name, "if") -> "IF-MIB"
String.starts_with?(base_name, "ipNetToMedia") -> "IP-MIB"
String.starts_with?(base_name, "ipNetToPhysical") -> "IP-MIB"
String.starts_with?(base_name, "dot1d") -> "BRIDGE-MIB"
String.starts_with?(base_name, "dot1q") -> "Q-BRIDGE-MIB"
true -> nil
end
# Map of prefix patterns to MIB modules
# Order matters: more specific prefixes before general ones
prefix_to_module = [
{"ifHC", "IF-MIB"},
{"ifIn", "IF-MIB"},
{"ifOut", "IF-MIB"},
{"if", "IF-MIB"},
{"ipNetToPhysical", "IP-MIB"},
{"ipNetToMedia", "IP-MIB"},
{"sys", "SNMPv2-MIB"},
{"dot1d", "BRIDGE-MIB"},
{"dot1q", "Q-BRIDGE-MIB"}
]
Enum.find_value(prefix_to_module, fn {prefix, module} ->
if String.starts_with?(base_name, prefix), do: module
end)
end
# Normalize name->oid map from arbitrary representations
@ -1164,35 +1178,36 @@ defmodule SnmpKit.SnmpMgr.MIB do
if Enum.all?(oid, &is_integer/1) do
{:ok, oid}
else
# Handle lists like [%{value: 1}, %{value: 3}, ...] possibly with names
vals =
Enum.map(oid, fn
%{value: v} when is_integer(v) ->
{:ok, v}
%{value: v} when is_binary(v) ->
case Integer.parse(v) do
{i, ""} -> {:ok, i}
_ -> :error
end
v when is_integer(v) ->
{:ok, v}
_ ->
:error
end)
if Enum.any?(vals, &(&1 == :error)) do
{:error, :unresolved_oid}
else
{:ok, Enum.map(vals, fn {:ok, i} -> i end)}
end
parse_complex_oid_elements(oid)
end
end
defp normalize_parsed_oid(_), do: {:error, :invalid_oid}
# Handle lists like [%{value: 1}, %{value: 3}, ...] possibly with names
defp parse_complex_oid_elements(oid) do
vals = Enum.map(oid, &normalize_oid_element/1)
if Enum.any?(vals, &(&1 == :error)) do
{:error, :unresolved_oid}
else
{:ok, Enum.map(vals, fn {:ok, i} -> i end)}
end
end
# Normalize a single OID element to an integer
defp normalize_oid_element(%{value: v}) when is_integer(v), do: {:ok, v}
defp normalize_oid_element(%{value: v}) when is_binary(v) do
case Integer.parse(v) do
{i, ""} -> {:ok, i}
_ -> :error
end
end
defp normalize_oid_element(v) when is_integer(v), do: {:ok, v}
defp normalize_oid_element(_), do: :error
defp load_mib_file_and_extract_mappings(mib_path) do
case File.read(mib_path) do
{:ok, mib_content} ->
@ -1209,76 +1224,104 @@ defmodule SnmpKit.SnmpMgr.MIB do
defp extract_mib_mappings(mib_data) do
# Extract name-to-OID mappings and basic metadata from parsed MIB data
definitions = Map.get(mib_data, :definitions, [])
# Build TC map first
tc_map =
definitions
|> Enum.filter(&(Map.get(&1, :__type__) == :textual_convention))
|> Enum.reduce(%{}, fn tc, acc ->
tc_name = Map.get(tc, :name)
tc_syntax = Map.get(tc, :syntax)
display_hint = Map.get(tc, :display_hint)
Map.put(acc, tc_name, %{syntax_base: syntax_base_from(tc_syntax), display_hint: display_hint})
end)
primitives =
MapSet.new([:integer, :octet_string, :object_identifier, :timeticks, :counter32, :counter64, :gauge32, :ip_address])
tc_map = build_tc_map(definitions)
{name_to_oid_map, name_to_meta} =
Enum.reduce(definitions, {%{}, %{}}, fn defn, {oid_acc, meta_acc} ->
case Map.get(defn, :__type__) do
:object_type ->
name = Map.get(defn, :name)
oid_any = Map.get(defn, :oid)
syntax_any = Map.get(defn, :syntax)
access = Map.get(defn, :max_access)
status = Map.get(defn, :status)
description = Map.get(defn, :description)
oid_acc2 =
case {name, normalize_parsed_oid(oid_any)} do
{name, {:ok, oid_list}} when is_binary(name) -> Map.put(oid_acc, name, oid_list)
_ -> oid_acc
end
{syntax_base, textual_convention, display_hint} =
case syntax_any do
# Named type referencing a TC like :DisplayString
t when is_atom(t) ->
if MapSet.member?(primitives, t) do
{syntax_base_from(syntax_any), textual_convention_from(syntax_any), nil}
else
tc_key = Atom.to_string(t)
case Map.get(tc_map, tc_key) do
%{syntax_base: base, display_hint: hint} -> {base, tc_key, hint}
_ -> {syntax_base_from(syntax_any), textual_convention_from(syntax_any), nil}
end
end
_ ->
{syntax_base_from(syntax_any), textual_convention_from(syntax_any), nil}
end
meta = %{
syntax_base: syntax_base,
textual_convention: textual_convention,
display_hint: display_hint,
access: access,
status: status,
description: description
}
{oid_acc2, Map.put(meta_acc, name, meta)}
_ ->
{oid_acc, meta_acc}
end
process_mib_definition(defn, oid_acc, meta_acc, tc_map)
end)
%{name_to_oid: name_to_oid_map, name_to_meta: name_to_meta}
end
# Build textual convention map from definitions
defp build_tc_map(definitions) do
definitions
|> Enum.filter(&(Map.get(&1, :__type__) == :textual_convention))
|> Enum.reduce(%{}, fn tc, acc ->
tc_name = Map.get(tc, :name)
tc_syntax = Map.get(tc, :syntax)
display_hint = Map.get(tc, :display_hint)
Map.put(acc, tc_name, %{syntax_base: syntax_base_from(tc_syntax), display_hint: display_hint})
end)
end
# Process a single MIB definition
defp process_mib_definition(%{__type__: :object_type} = defn, oid_acc, meta_acc, tc_map) do
name = Map.get(defn, :name)
oid_any = Map.get(defn, :oid)
oid_acc2 = maybe_add_oid_mapping(name, oid_any, oid_acc)
meta = build_object_type_meta(defn, tc_map)
{oid_acc2, Map.put(meta_acc, name, meta)}
end
defp process_mib_definition(_defn, oid_acc, meta_acc, _tc_map), do: {oid_acc, meta_acc}
# Add OID mapping if valid
defp maybe_add_oid_mapping(name, oid_any, oid_acc) when is_binary(name) do
case normalize_parsed_oid(oid_any) do
{:ok, oid_list} -> Map.put(oid_acc, name, oid_list)
_ -> oid_acc
end
end
defp maybe_add_oid_mapping(_name, _oid_any, oid_acc), do: oid_acc
# Build metadata for an object type definition
defp build_object_type_meta(defn, tc_map) do
syntax_any = Map.get(defn, :syntax)
{syntax_base, textual_convention, display_hint} = resolve_syntax_metadata(syntax_any, tc_map)
%{
syntax_base: syntax_base,
textual_convention: textual_convention,
display_hint: display_hint,
access: Map.get(defn, :max_access),
status: Map.get(defn, :status),
description: Map.get(defn, :description)
}
end
# Resolve syntax metadata, looking up textual conventions
defp resolve_syntax_metadata(syntax_any, tc_map) when is_atom(syntax_any) do
primitives =
MapSet.new([
:integer,
:octet_string,
:object_identifier,
:timeticks,
:counter32,
:counter64,
:gauge32,
:ip_address
])
if MapSet.member?(primitives, syntax_any) do
{syntax_base_from(syntax_any), textual_convention_from(syntax_any), nil}
else
lookup_tc_metadata(syntax_any, tc_map)
end
end
defp resolve_syntax_metadata(syntax_any, _tc_map) do
{syntax_base_from(syntax_any), textual_convention_from(syntax_any), nil}
end
# Look up textual convention metadata
defp lookup_tc_metadata(syntax_atom, tc_map) do
tc_key = Atom.to_string(syntax_atom)
case Map.get(tc_map, tc_key) do
%{syntax_base: base, display_hint: hint} ->
{base, tc_key, hint}
_ ->
{syntax_base_from(syntax_atom), textual_convention_from(syntax_atom), nil}
end
end
defp merge_mib_data(state, _mib_data) do
# This would merge the new MIB data with existing state
# For now, just return the current state