towerops/lib/snmpkit/snmp_lib/oid.ex
Graham McIntire 348975dcc9 Rewrite polling_offset, changelog_parser, and user_agent_parser in Gleam (#103)
Migrate three pure-function modules from Elixir to Gleam:

- polling_offset: deterministic hash-based polling offset (pure Gleam, no FFI)
- changelog_parser: text parsing logic in Gleam, thin Elixir wrapper for File I/O
- user_agent_parser: regex-based UA parsing in Gleam with Erlang FFI for
  atom-keyed map conversion

Add gleam_regexp dependency for regex support in Gleam modules.
Update all call sites and tests to use Gleam module atoms.

Reviewed-on: graham/towerops-web#103
2026-03-21 10:11:06 -05:00

406 lines
12 KiB
Elixir

defmodule SnmpKit.SnmpLib.OID do
@moduledoc """
Comprehensive OID (Object Identifier) manipulation utilities for SNMP operations.
Core logic implemented in Gleam (`src/snmpkit/snmp_lib/oid.gleam`).
This module provides the Elixir API with type guards and polymorphic dispatch.
## Features
- String/list format conversions with validation
- Support for both OID formats: "1.3.6.1.2.1.1" and ".1.3.6.1.2.1.1"
- OID tree operations (parent/child relationships)
- SNMP table index parsing and construction
- OID comparison and sorting
- Enterprise OID utilities
## Examples
{:ok, oid_list} = SnmpKit.SnmpLib.OID.string_to_list("1.3.6.1.2.1.1.1.0")
{:ok, oid_string} = SnmpKit.SnmpLib.OID.list_to_string([1, 3, 6, 1, 2, 1, 1, 1, 0])
true = SnmpKit.SnmpLib.OID.child_of?([1, 3, 6, 1, 2, 1, 1, 1, 0], [1, 3, 6, 1, 2, 1])
:lt = SnmpKit.SnmpLib.OID.compare([1, 3, 6, 1], [1, 3, 6, 2])
"""
@gleam :snmpkit@snmp_lib@oid
@type oid :: [non_neg_integer()]
@type oid_string :: String.t()
@type table_oid :: oid()
@type index :: [non_neg_integer()]
## String/List Conversions
@doc """
Converts an OID string to a list of integers.
## Examples
iex> SnmpKit.SnmpLib.OID.string_to_list("1.3.6.1.2.1.1.1.0")
{:ok, [1, 3, 6, 1, 2, 1, 1, 1, 0]}
iex> SnmpKit.SnmpLib.OID.string_to_list(".1.3.6.1.2.1.1.1.0")
{:ok, [1, 3, 6, 1, 2, 1, 1, 1, 0]}
iex> SnmpKit.SnmpLib.OID.string_to_list("1.3.6")
{:ok, [1, 3, 6]}
iex> SnmpKit.SnmpLib.OID.string_to_list(".1.3.6")
{:ok, [1, 3, 6]}
iex> SnmpKit.SnmpLib.OID.string_to_list("")
{:error, :empty_oid}
iex> SnmpKit.SnmpLib.OID.string_to_list("1.3.6.1.a.2")
{:error, :invalid_oid_string}
iex> SnmpKit.SnmpLib.OID.string_to_list("1.3.6.1.2.-1")
{:error, :invalid_oid_string}
"""
@spec string_to_list(oid_string()) :: {:ok, oid()} | {:error, atom()}
def string_to_list(oid_string) when is_binary(oid_string) do
@gleam.string_to_list(oid_string)
end
def string_to_list(_), do: {:error, :invalid_input}
@doc """
Converts an OID list to a dot-separated string.
## Examples
iex> SnmpKit.SnmpLib.OID.list_to_string([1, 3, 6, 1, 2, 1, 1, 1, 0])
{:ok, "1.3.6.1.2.1.1.1.0"}
"""
@spec list_to_string(oid()) :: {:ok, oid_string()} | {:error, atom()}
def list_to_string(oid_list) when is_list(oid_list) do
@gleam.list_to_string(oid_list)
end
def list_to_string(_), do: {:error, :invalid_input}
## Tree Operations
@doc """
Checks if one OID is a child of another.
## Examples
iex> SnmpKit.SnmpLib.OID.child_of?([1, 3, 6, 1, 2, 1, 1, 1, 0], [1, 3, 6, 1, 2, 1])
true
"""
@spec child_of?(oid(), oid()) :: boolean()
def child_of?(child_oid, parent_oid) when is_list(child_oid) and is_list(parent_oid) do
@gleam.child_of(child_oid, parent_oid)
end
def child_of?(_, _), do: false
@doc """
Checks if one OID is a parent of another.
"""
@spec parent_of?(oid(), oid()) :: boolean()
def parent_of?(parent_oid, child_oid), do: child_of?(child_oid, parent_oid)
@doc """
Gets the parent OID by removing the last component.
## Examples
iex> SnmpKit.SnmpLib.OID.get_parent([1, 3, 6, 1, 2, 1, 1, 1, 0])
{:ok, [1, 3, 6, 1, 2, 1, 1, 1]}
iex> SnmpKit.SnmpLib.OID.get_parent([])
{:error, :root_oid}
iex> SnmpKit.SnmpLib.OID.get_parent([1])
{:error, :root_oid}
"""
@spec get_parent(oid()) :: {:ok, oid()} | {:error, atom()}
def get_parent(oid) when is_list(oid), do: @gleam.get_parent(oid)
def get_parent(_), do: {:error, :invalid_input}
@doc """
Gets the immediate children prefix for an OID in a given set.
"""
@spec get_children(oid(), [oid()]) :: [oid()]
def get_children(parent_oid, oid_set) when is_list(parent_oid) and is_list(oid_set) do
@gleam.get_children(parent_oid, oid_set)
end
def get_children(_, _), do: []
@doc """
Get standard SNMP OID prefixes.
## Examples
iex> SnmpKit.SnmpLib.OID.standard_prefix(:internet)
[1, 3, 6, 1]
iex> SnmpKit.SnmpLib.OID.standard_prefix(:mgmt)
[1, 3, 6, 1, 2]
"""
@spec standard_prefix(atom()) :: oid() | nil
def standard_prefix(:internet), do: [1, 3, 6, 1]
def standard_prefix(:mgmt), do: [1, 3, 6, 1, 2]
def standard_prefix(:mib_2), do: [1, 3, 6, 1, 2, 1]
def standard_prefix(:enterprises), do: [1, 3, 6, 1, 4, 1]
def standard_prefix(:experimental), do: [1, 3, 6, 1, 3]
def standard_prefix(:private), do: [1, 3, 6, 1, 4]
def standard_prefix(_), do: nil
@doc """
Gets the next OID in lexicographic order from a given set.
"""
@spec get_next_oid(oid(), [oid()]) :: {:ok, oid()} | {:error, atom()}
def get_next_oid(current_oid, oid_set) when is_list(current_oid) and is_list(oid_set) do
@gleam.get_next_oid(current_oid, oid_set)
end
def get_next_oid(_, _), do: {:error, :invalid_input}
## Comparison Operations
@doc """
Compares two OIDs lexicographically.
## Examples
iex> SnmpKit.SnmpLib.OID.compare([1, 3, 6, 1], [1, 3, 6, 2])
:lt
iex> SnmpKit.SnmpLib.OID.compare([1, 3, 6, 1], [1, 3, 6, 1])
:eq
iex> SnmpKit.SnmpLib.OID.compare([1, 3, 6, 2], [1, 3, 6, 1])
:gt
"""
@spec compare(oid(), oid()) :: :lt | :eq | :gt
def compare(oid1, oid2) when is_list(oid1) and is_list(oid2) do
@gleam.compare_oids(oid1, oid2)
end
@doc """
Sorts a list of OIDs in lexicographic order.
"""
@spec sort([oid()]) :: [oid()]
def sort(oid_list) when is_list(oid_list), do: @gleam.sort_oids(oid_list)
def sort(_), do: []
## Table Operations
@doc """
Extracts table index from an instance OID given the table column OID.
## Examples
iex> SnmpKit.SnmpLib.OID.extract_table_index([1, 3, 6, 1, 2, 1, 2, 2, 1, 1], [1, 3, 6, 1, 2, 1, 2, 2, 1, 1, 1])
{:ok, [1]}
"""
@spec extract_table_index(table_oid(), oid()) :: {:ok, index()} | {:error, atom()}
def extract_table_index(table_oid, instance_oid) when is_list(table_oid) and is_list(instance_oid) do
@gleam.extract_table_index(table_oid, instance_oid)
end
def extract_table_index(_, _), do: {:error, :invalid_input}
@doc """
Builds a table instance OID from table OID and index.
"""
@spec build_table_instance(table_oid(), index()) :: {:ok, oid()} | {:error, atom()}
def build_table_instance(table_oid, index) when is_list(table_oid) and is_list(index) do
@gleam.build_table_instance(table_oid, index)
end
def build_table_instance(_, _), do: {:error, :invalid_input}
@doc """
Parses a table index according to index syntax definition.
## Examples
iex> SnmpKit.SnmpLib.OID.parse_table_index([42], :integer)
{:ok, 42}
iex> SnmpKit.SnmpLib.OID.parse_table_index([4, 116, 101, 115, 116], {:variable_string})
{:ok, "test"}
"""
@spec parse_table_index(index(), term()) :: {:ok, term()} | {:error, atom()}
def parse_table_index(index, :integer) when is_list(index) do
@gleam.parse_integer_index(index)
end
def parse_table_index(index, {:string, length}) when is_list(index) do
@gleam.parse_fixed_string_index(index, length)
end
def parse_table_index(index, {:variable_string}) when is_list(index) do
@gleam.parse_variable_string_index(index)
end
def parse_table_index(index, syntax_list) when is_list(index) and is_list(syntax_list) do
case parse_index_components(index, syntax_list, []) do
{:ok, {parsed, _remaining}} -> {:ok, parsed}
{:error, reason} -> {:error, reason}
end
end
def parse_table_index(_, _), do: {:error, :unsupported_syntax}
@doc """
Builds a table index from parsed components according to syntax.
## Examples
iex> SnmpKit.SnmpLib.OID.build_table_index(42, :integer)
{:ok, [42]}
iex> SnmpKit.SnmpLib.OID.build_table_index("test", {:variable_string})
{:ok, [4, 116, 101, 115, 116]}
"""
@spec build_table_index(term(), term()) :: {:ok, index()} | {:error, atom()}
def build_table_index(values, syntax_list) when is_list(values) and is_list(syntax_list) do
if length(values) == length(syntax_list) do
build_compound_index(values, syntax_list)
else
{:error, :syntax_value_mismatch}
end
end
def build_table_index(value, :integer) when is_integer(value) and value >= 0 do
@gleam.build_integer_index(value)
end
def build_table_index(value, {:string, length}) when is_binary(value) do
@gleam.build_fixed_string_index(value, length)
end
def build_table_index(value, {:variable_string}) when is_binary(value) do
@gleam.build_variable_string_index(value)
end
def build_table_index(_, _), do: {:error, :unsupported_syntax}
## Validation
@doc """
Validates an OID list for correctness.
## Examples
iex> SnmpKit.SnmpLib.OID.valid_oid?([1, 3, 6, 1, 2, 1, 1, 1, 0])
:ok
iex> SnmpKit.SnmpLib.OID.valid_oid?([])
{:error, :empty_oid}
"""
@spec valid_oid?(oid()) :: :ok | {:error, atom()}
def valid_oid?(oid) when is_list(oid) do
case @gleam.validate(oid) do
{:ok, _} -> :ok
error -> error
end
end
def valid_oid?(_), do: {:error, :invalid_input}
@doc """
Normalizes an OID to a consistent format (list).
## Examples
iex> SnmpKit.SnmpLib.OID.normalize("1.3.6.1")
{:ok, [1, 3, 6, 1]}
iex> SnmpKit.SnmpLib.OID.normalize([1, 3, 6, 1])
{:ok, [1, 3, 6, 1]}
"""
@spec normalize(oid() | oid_string()) :: {:ok, oid()} | {:error, atom()}
def normalize(oid) when is_list(oid), do: @gleam.normalize_list(oid)
def normalize(oid) when is_binary(oid), do: @gleam.normalize_string(oid)
def normalize(_), do: {:error, :invalid_input}
## Standard OID Utilities
@spec mib_2() :: oid()
def mib_2, do: @gleam.mib_2()
@spec enterprises() :: oid()
def enterprises, do: @gleam.enterprises()
@spec experimental() :: oid()
def experimental, do: @gleam.experimental()
@spec private() :: oid()
def private, do: @gleam.private()
@spec mib_2?(oid()) :: boolean()
def mib_2?(oid), do: @gleam.is_mib_2(oid)
@spec enterprise?(oid()) :: boolean()
def enterprise?(oid), do: @gleam.is_enterprise(oid)
@spec experimental?(oid()) :: boolean()
def experimental?(oid), do: @gleam.is_experimental(oid)
@spec private?(oid()) :: boolean()
def private?(oid), do: @gleam.is_private(oid)
@doc """
Gets the enterprise number from an enterprise OID.
## Examples
iex> SnmpKit.SnmpLib.OID.get_enterprise_number([1, 3, 6, 1, 4, 1, 9, 1, 1])
{:ok, 9}
"""
@spec get_enterprise_number(oid()) :: {:ok, non_neg_integer()} | {:error, atom()}
def get_enterprise_number(oid) when is_list(oid), do: @gleam.get_enterprise_number(oid)
def get_enterprise_number(_), do: {:error, :invalid_input}
## Private Helpers for compound table index operations
defp parse_index_components(index, [], acc) do
{:ok, {Enum.reverse(acc), index}}
end
defp parse_index_components(index, [syntax | rest_syntax], acc) do
case parse_table_index(index, syntax) do
{:ok, value} ->
case build_table_index(value, syntax) do
{:ok, consumed_index} ->
consumed_length = length(consumed_index)
remaining_index = Enum.drop(index, consumed_length)
parse_index_components(remaining_index, rest_syntax, [value | acc])
{:error, reason} ->
{:error, reason}
end
{:error, reason} ->
{:error, reason}
end
end
defp build_compound_index(values, syntax_list) do
index_parts =
values
|> Enum.zip(syntax_list)
|> Enum.map(fn {val, syntax} ->
case build_table_index(val, syntax) do
{:ok, idx} -> idx
{:error, reason} -> {:error, reason}
end
end)
case Enum.find(index_parts, &match?({:error, _}, &1)) do
nil ->
index = List.flatten(index_parts)
{:ok, index}
{:error, reason} ->
{:error, reason}
end
end
end