Graham McIntire
efaf5558ff
Phase 1: Foundation Types (100% complete)
- query_helpers: SQL LIKE sanitization with pipe operators
- numeric: Integer parsing with pattern matching guards
- result: Pure Elixir Result monad (map, and_then, unwrap_or)
Phase 2: Ecto Domain Types (100% complete)
- ip_address: IPv4/IPv6 validation using :inet directly
- mac_address: Multi-format MAC parsing (colon/hyphen/dot/compact)
- snmp_oid: OID parsing/manipulation with recursive pattern matching
All 198 tests passing across converted modules.
API changed from Gleam-style {:some/:none to idiomatic {:ok/:error.
Refactored parse_numeric_oid to use with statement, reducing nesting depth.
Reviewed-on: graham/towerops-web#196
306 lines
9.3 KiB
Elixir
306 lines
9.3 KiB
Elixir
defmodule SnmpKit.SnmpLib.PDU.Constants do
|
|
@moduledoc """
|
|
Constants and type definitions for SNMP PDU operations.
|
|
|
|
Provides pure computation (version normalization, OID parsing, message flag encoding),
|
|
type definitions, atom-based lookups, and binary encodings.
|
|
"""
|
|
|
|
import Bitwise
|
|
|
|
# Type definitions
|
|
@type snmp_type ::
|
|
:integer
|
|
| :octet_string
|
|
| :null
|
|
| :object_identifier
|
|
| :counter32
|
|
| :gauge32
|
|
| :timeticks
|
|
| :counter64
|
|
| :ip_address
|
|
| :opaque_type
|
|
| :no_such_object
|
|
| :no_such_instance
|
|
| :end_of_mib_view
|
|
|
|
# SNMP PDU Types
|
|
@get_request 0xA0
|
|
@getnext_request 0xA1
|
|
@get_response 0xA2
|
|
@set_request 0xA3
|
|
@getbulk_request 0xA5
|
|
|
|
# SNMP Data Types
|
|
@integer 0x02
|
|
@octet_string 0x04
|
|
@null 0x05
|
|
@object_identifier 0x06
|
|
@counter32 0x41
|
|
@gauge32 0x42
|
|
@timeticks 0x43
|
|
@counter64 0x46
|
|
@ip_address 0x40
|
|
@opaque_type 0x44
|
|
@no_such_object 0x80
|
|
@no_such_instance 0x81
|
|
@end_of_mib_view 0x82
|
|
|
|
# SNMP Error Status Codes
|
|
@no_error 0
|
|
@too_big 1
|
|
@no_such_name 2
|
|
@bad_value 3
|
|
@read_only 4
|
|
@gen_err 5
|
|
|
|
@type snmp_version :: :v1 | :v2c | :v2 | :v3 | 0 | 1 | 3
|
|
@type pdu_type ::
|
|
:get_request | :get_next_request | :get_response | :set_request | :get_bulk_request
|
|
@type error_status :: 0..18
|
|
@type oid :: [non_neg_integer()] | binary()
|
|
@type snmp_value :: any()
|
|
@type varbind :: {oid(), atom(), snmp_value()}
|
|
|
|
@type base_pdu :: %{
|
|
type: pdu_type(),
|
|
request_id: non_neg_integer(),
|
|
error_status: error_status(),
|
|
error_index: non_neg_integer(),
|
|
varbinds: [varbind()]
|
|
}
|
|
|
|
@type bulk_pdu :: %{
|
|
type: :get_bulk_request,
|
|
request_id: non_neg_integer(),
|
|
error_status: error_status(),
|
|
error_index: non_neg_integer(),
|
|
varbinds: [varbind()],
|
|
non_repeaters: non_neg_integer(),
|
|
max_repetitions: non_neg_integer()
|
|
}
|
|
|
|
@type pdu :: base_pdu() | bulk_pdu()
|
|
|
|
# SNMPv1/v2c message format
|
|
@type v1v2c_message :: %{
|
|
version: snmp_version() | non_neg_integer(),
|
|
community: binary(),
|
|
pdu: pdu()
|
|
}
|
|
|
|
# SNMPv3 message format
|
|
@type v3_message :: %{
|
|
version: 3,
|
|
msg_id: non_neg_integer(),
|
|
msg_max_size: non_neg_integer(),
|
|
msg_flags: binary(),
|
|
msg_security_model: non_neg_integer(),
|
|
msg_security_parameters: binary(),
|
|
msg_data: scoped_pdu()
|
|
}
|
|
|
|
@type message :: v1v2c_message() | v3_message()
|
|
|
|
# SNMPv3 specific types
|
|
@type scoped_pdu :: %{
|
|
context_engine_id: binary(),
|
|
context_name: binary(),
|
|
pdu: pdu()
|
|
}
|
|
|
|
@type msg_flags :: %{
|
|
auth: boolean(),
|
|
priv: boolean(),
|
|
reportable: boolean()
|
|
}
|
|
|
|
# Error status code accessors
|
|
def no_error, do: @no_error
|
|
def too_big, do: @too_big
|
|
def no_such_name, do: @no_such_name
|
|
def bad_value, do: @bad_value
|
|
def read_only, do: @read_only
|
|
def gen_err, do: @gen_err
|
|
|
|
# SNMPv3 constants
|
|
@usm_security_model 3
|
|
@default_max_message_size 65_507
|
|
|
|
def usm_security_model, do: @usm_security_model
|
|
def default_max_message_size, do: @default_max_message_size
|
|
|
|
# PDU type constants accessors
|
|
def get_request, do: @get_request
|
|
def getnext_request, do: @getnext_request
|
|
def get_response, do: @get_response
|
|
def set_request, do: @set_request
|
|
def getbulk_request, do: @getbulk_request
|
|
|
|
# Data type constants accessors
|
|
def integer, do: @integer
|
|
def octet_string, do: @octet_string
|
|
def null, do: @null
|
|
def object_identifier, do: @object_identifier
|
|
def counter32, do: @counter32
|
|
def gauge32, do: @gauge32
|
|
def timeticks, do: @timeticks
|
|
def counter64, do: @counter64
|
|
def ip_address, do: @ip_address
|
|
def opaque_type, do: @opaque_type
|
|
def no_such_object, do: @no_such_object
|
|
def no_such_instance, do: @no_such_instance
|
|
def end_of_mib_view, do: @end_of_mib_view
|
|
|
|
# Version normalization
|
|
def normalize_version(:v1), do: 0
|
|
def normalize_version(:v2c), do: 1
|
|
def normalize_version(:v2), do: 1
|
|
def normalize_version(:v3), do: 3
|
|
def normalize_version(v) when is_integer(v), do: v
|
|
def normalize_version(_), do: 0
|
|
|
|
# OID normalization
|
|
def normalize_oid(oid) when is_list(oid), do: oid
|
|
def normalize_oid(oid) when is_binary(oid), do: normalize_oid_string(oid)
|
|
def normalize_oid(_), do: [1, 3, 6, 1]
|
|
|
|
defp normalize_oid_string(oid) do
|
|
default_oid = [1, 3, 6, 1]
|
|
|
|
parts = String.split(oid, ".")
|
|
|
|
case parts do
|
|
[] ->
|
|
default_oid
|
|
|
|
_ ->
|
|
case parse_all_oid_parts(parts) do
|
|
{:ok, parsed} -> parsed
|
|
{:error, _} -> default_oid
|
|
end
|
|
end
|
|
end
|
|
|
|
defp parse_all_oid_parts(parts) do
|
|
Enum.reduce_while(parts, {:ok, []}, fn part, {:ok, acc} ->
|
|
case Integer.parse(part) do
|
|
{num, ""} when num >= 0 -> {:cont, {:ok, acc ++ [num]}}
|
|
_ -> {:halt, {:error, nil}}
|
|
end
|
|
end)
|
|
end
|
|
|
|
# New utility functions for type conversion
|
|
def pdu_type_to_tag(:get_request), do: @get_request
|
|
def pdu_type_to_tag(:get_next_request), do: @getnext_request
|
|
def pdu_type_to_tag(:get_response), do: @get_response
|
|
def pdu_type_to_tag(:set_request), do: @set_request
|
|
def pdu_type_to_tag(:get_bulk_request), do: @getbulk_request
|
|
|
|
def tag_to_pdu_type(@get_request), do: :get_request
|
|
def tag_to_pdu_type(@getnext_request), do: :get_next_request
|
|
def tag_to_pdu_type(@get_response), do: :get_response
|
|
def tag_to_pdu_type(@set_request), do: :set_request
|
|
def tag_to_pdu_type(@getbulk_request), do: :get_bulk_request
|
|
def tag_to_pdu_type(_), do: nil
|
|
|
|
def data_type_to_tag(:integer), do: @integer
|
|
def data_type_to_tag(:octet_string), do: @octet_string
|
|
def data_type_to_tag(:null), do: @null
|
|
def data_type_to_tag(:object_identifier), do: @object_identifier
|
|
def data_type_to_tag(:counter32), do: @counter32
|
|
def data_type_to_tag(:gauge32), do: @gauge32
|
|
def data_type_to_tag(:timeticks), do: @timeticks
|
|
def data_type_to_tag(:counter64), do: @counter64
|
|
def data_type_to_tag(:ip_address), do: @ip_address
|
|
def data_type_to_tag(:opaque), do: @opaque_type
|
|
def data_type_to_tag(:no_such_object), do: @no_such_object
|
|
def data_type_to_tag(:no_such_instance), do: @no_such_instance
|
|
def data_type_to_tag(:end_of_mib_view), do: @end_of_mib_view
|
|
|
|
def tag_to_data_type(@integer), do: :integer
|
|
def tag_to_data_type(@octet_string), do: :octet_string
|
|
def tag_to_data_type(@null), do: :null
|
|
def tag_to_data_type(@object_identifier), do: :object_identifier
|
|
def tag_to_data_type(@counter32), do: :counter32
|
|
def tag_to_data_type(@gauge32), do: :gauge32
|
|
def tag_to_data_type(@timeticks), do: :timeticks
|
|
def tag_to_data_type(@counter64), do: :counter64
|
|
def tag_to_data_type(@ip_address), do: :ip_address
|
|
def tag_to_data_type(@opaque_type), do: :opaque
|
|
def tag_to_data_type(@no_such_object), do: :no_such_object
|
|
def tag_to_data_type(@no_such_instance), do: :no_such_instance
|
|
def tag_to_data_type(@end_of_mib_view), do: :end_of_mib_view
|
|
def tag_to_data_type(_), do: nil
|
|
|
|
@doc """
|
|
Normalizes an SNMP type atom.
|
|
"""
|
|
@spec normalize_type(atom()) :: snmp_type()
|
|
def normalize_type(:string), do: :octet_string
|
|
def normalize_type(type), do: type
|
|
|
|
@doc """
|
|
Converts an error status atom to its numeric code.
|
|
"""
|
|
@spec error_status_to_code(atom()) :: non_neg_integer()
|
|
def error_status_to_code(:no_error), do: @no_error
|
|
def error_status_to_code(:too_big), do: @too_big
|
|
def error_status_to_code(:no_such_name), do: @no_such_name
|
|
def error_status_to_code(:bad_value), do: @bad_value
|
|
def error_status_to_code(:read_only), do: @read_only
|
|
def error_status_to_code(:gen_err), do: @gen_err
|
|
def error_status_to_code(code) when is_integer(code), do: code
|
|
|
|
@doc """
|
|
Converts an error status code to its atom representation.
|
|
"""
|
|
@spec error_status_to_atom(non_neg_integer()) :: atom()
|
|
def error_status_to_atom(@no_error), do: :no_error
|
|
def error_status_to_atom(@too_big), do: :too_big
|
|
def error_status_to_atom(@no_such_name), do: :no_such_name
|
|
def error_status_to_atom(@bad_value), do: :bad_value
|
|
def error_status_to_atom(@read_only), do: :read_only
|
|
def error_status_to_atom(@gen_err), do: :gen_err
|
|
def error_status_to_atom(code), do: code
|
|
|
|
@doc """
|
|
Encodes SNMPv3 message flags to binary format.
|
|
"""
|
|
@spec encode_msg_flags(msg_flags()) :: binary()
|
|
def encode_msg_flags(%{auth: auth, priv: priv, reportable: reportable}) do
|
|
flags = 0
|
|
flags = if auth, do: flags ||| 1, else: flags
|
|
flags = if priv, do: flags ||| 2, else: flags
|
|
flags = if reportable, do: flags ||| 4, else: flags
|
|
<<flags>>
|
|
end
|
|
|
|
@doc """
|
|
Decodes SNMPv3 message flags from binary format.
|
|
"""
|
|
@spec decode_msg_flags(binary()) :: msg_flags()
|
|
def decode_msg_flags(<<flags::8>>) do
|
|
auth = (flags &&& 1) != 0
|
|
priv = (flags &&& 2) != 0
|
|
reportable = (flags &&& 4) != 0
|
|
%{auth: auth, priv: priv, reportable: reportable}
|
|
end
|
|
|
|
@doc """
|
|
Creates default SNMPv3 message flags for a security level.
|
|
"""
|
|
@spec default_msg_flags(atom()) :: msg_flags()
|
|
def default_msg_flags(:no_auth_no_priv) do
|
|
%{auth: false, priv: false, reportable: true}
|
|
end
|
|
|
|
def default_msg_flags(:auth_no_priv) do
|
|
%{auth: true, priv: false, reportable: true}
|
|
end
|
|
|
|
def default_msg_flags(:auth_priv) do
|
|
%{auth: true, priv: true, reportable: true}
|
|
end
|
|
end
|