SnmpKit.SnmpLib.Security.USM (snmpkit v1.4.0)

User Security Model (USM) implementation for SNMPv3 - RFC 3414 compliant.

The User Security Model provides the foundation for SNMPv3 security by implementing:

  • User-based authentication with multiple protocols
  • Privacy (encryption) for message confidentiality
  • Time synchronization to prevent replay attacks
  • Engine discovery for secure agent communication
  • Security parameter validation and error handling

RFC 3414 Compliance

This implementation fully complies with RFC 3414 "User-based Security Model (USM) for version 3 of the Simple Network Management Protocol (SNMPv3)" including:

  • Message authentication using HMAC-MD5 and HMAC-SHA
  • Privacy using DES and AES encryption
  • Key derivation using password localization
  • Time window validation for message freshness
  • Engine ID discovery and management

Architecture

The USM coordinates with other security modules:

SnmpKit.SnmpLib.Security.USM
├── Auth protocols (MD5, SHA variants)
├── Priv protocols (DES, AES variants)
├── Key derivation and management
└── Engine and time management

Usage Examples

Engine Discovery

# Discover remote engine for secure communication
{:ok, engine_id} = SnmpKit.SnmpLib.Security.USM.discover_engine("192.168.1.1")

# Time synchronization
{:ok, {boots, time}} = SnmpKit.SnmpLib.Security.USM.synchronize_time("192.168.1.1", engine_id)

Message Processing

# Process outgoing secure message
{:ok, secure_message} = SnmpKit.SnmpLib.Security.USM.process_outgoing_message(
  user, message, security_level
)

# Process incoming secure message
{:ok, {plain_message, user}} = SnmpKit.SnmpLib.Security.USM.process_incoming_message(
  secure_message, user_database
)

Security Considerations

  • Engine boot counters must be persistent across restarts
  • Time synchronization is critical for security
  • Failed authentication attempts should be logged
  • Key material should never be logged or persisted in plain text

Summary

Types

engine_boots()
engine_id()
engine_time()
message_flags()
security_level()
security_name()
security_parameters()
user_entry()

Functions

discover_engine(host, opts \\ [])

Discovers the engine ID of a remote SNMP agent.

generate_error_report(error_type, context)

Generates security error reports for invalid messages.

get_engine_time(boot_timestamp)

Calculates current engine time since boot.

increment_engine_boots(current_boots)

Updates engine boot counter, handling rollover at maximum value.

process_incoming_message(secure_message, user_database, opts \\ [])

Processes an incoming SNMP message with USM security.

process_outgoing_message(user, message, security_level)

Processes an outgoing SNMP message with USM security.

synchronize_time(host, engine_id, opts \\ [])

Synchronizes time with a remote SNMP agent.

validate_time_window(local_boots, local_time, remote_boots, remote_time, opts \\ [])

Validates time-based security parameters to prevent replay attacks.

Types

engine_boots()

@type engine_boots() :: non_neg_integer()

engine_id()

@type engine_id() :: binary()

engine_time()

@type engine_time() :: non_neg_integer()

message_flags()

@type message_flags() :: %{
  auth_flag: boolean(),
  priv_flag: boolean(),
  reportable_flag: boolean()
}

security_level()

@type security_level() :: :no_auth_no_priv | :auth_no_priv | :auth_priv

security_name()

@type security_name() :: binary()

security_parameters()

@type security_parameters() :: %{
  authoritative_engine_id: engine_id(),
  authoritative_engine_boots: engine_boots(),
  authoritative_engine_time: engine_time(),
  user_name: security_name(),
  authentication_parameters: binary(),
  privacy_parameters: binary()
}

user_entry()

@type user_entry() :: %{
  security_name: security_name(),
  auth_protocol: atom(),
  priv_protocol: atom(),
  auth_key: binary(),
  priv_key: binary(),
  engine_id: engine_id()
}

Functions

discover_engine(host, opts \\ [])

@spec discover_engine(
  binary(),
  keyword()
) :: {:ok, engine_id()} | {:error, atom()}

Discovers the engine ID of a remote SNMP agent.

Engine discovery is the first step in establishing secure communication with a remote SNMPv3 agent. This function sends a discovery request and retrieves the agent's authoritative engine ID.

Parameters

  • host: Target agent IP address or hostname
  • opts: Discovery options including port, timeout, and community

Returns

  • {:ok, engine_id}: Successfully discovered engine ID
  • {:error, reason}: Discovery failed

Examples

{:ok, engine_id} = SnmpKit.SnmpLib.Security.USM.discover_engine("192.168.1.1")
{:ok, engine_id} = SnmpKit.SnmpLib.Security.USM.discover_engine("10.0.0.1", port: 1161, timeout: 5000)

generate_error_report(error_type, context)

@spec generate_error_report(atom(), map()) :: {:ok, binary()} | {:error, atom()}

Generates security error reports for invalid messages.

USM error reports are sent back to the originator to indicate security violations or configuration issues.

get_engine_time(boot_timestamp)

@spec get_engine_time(non_neg_integer()) :: engine_time()

Calculates current engine time since boot.

increment_engine_boots(current_boots)

@spec increment_engine_boots(engine_boots()) :: engine_boots()

Updates engine boot counter, handling rollover at maximum value.

process_incoming_message(secure_message, user_database, opts \\ [])

@spec process_incoming_message(binary(), map(), keyword()) ::
  {:ok, {map(), user_entry()}} | {:error, atom()}

Processes an incoming SNMP message with USM security.

Implements the receiving side of RFC 3414 section 3.2: the header is read to find msgUserName, the user is looked up in user_database (a map from security name to user entry), the engine ID, security level and timeliness window are checked, and finally the message is authenticated and decrypted with that user's keys.

User entries must carry the authoritative engine's current engine_boots and engine_time; those are what incoming messages are checked against.

Options

  • :time_window - allowed |time difference| in seconds (default 150)

Returns

  • {:ok, {message, user}} where message is the decoded SNMPv3 message (see SnmpKit.SnmpLib.PDU.V3Encoder.decode_message/2)
  • {:error, reason} with the RFC 3414 failure class, e.g. :unknown_user_name, :unknown_engine_id, :unsupported_security_level, :not_in_time_window, :authentication_mismatch

process_outgoing_message(user, message, security_level)

@spec process_outgoing_message(user_entry(), binary(), security_level()) ::
  {:ok, binary()} | {:error, atom()}

Processes an outgoing SNMP message with USM security.

This function applies authentication and/or privacy protection to an outgoing message based on the user's security level configuration.

synchronize_time(host, engine_id, opts \\ [])

@spec synchronize_time(binary(), engine_id(), keyword()) ::
  {:ok, {engine_boots(), engine_time()}} | {:error, atom()}

Synchronizes time with a remote SNMP agent.

Time synchronization is required for authenticated communication to prevent replay attacks. This function retrieves the agent's current boot counter and engine time (RFC 3414 section 2.3).

The request is sent with msgAuthoritativeEngineBoots/Time of zero. The agent answers with a Report PDU (usmStatsNotInTimeWindows) whose security parameters carry its real boots and time, which are returned.

Options

  • :user - a security user (see SnmpKit.SnmpLib.Security.create_user/2) with valid authentication keys. Agents only answer time-sync requests they can authenticate, so this is needed for real devices.
  • :security_name, :port, :timeout

validate_time_window(local_boots, local_time, remote_boots, remote_time, opts \\ [])

@spec validate_time_window(
  engine_boots(),
  engine_time(),
  engine_boots(),
  engine_time(),
  keyword()
) :: :ok | {:error, atom()}

Validates time-based security parameters to prevent replay attacks.

Per RFC 3414, messages are considered fresh if:

  • Engine boots match (within 1)
  • Engine time is within 150 seconds