RouterOS API

CIHex.pmDocumentation

Elixir client for MikroTik RouterOS binary API. Supports both plain TCP (port 8728) and TLS (port 8729) connections.

Features

Installation

Add routeros_api to your list of dependencies in mix.exs:

def deps do
  [
    {:routeros_api, "~> 0.3.0"}
  ]
end

Or from GitHub:

def deps do
  [
    {:routeros_api, github: "jlbyh2o/routeros_api"}
  ]
end

Quick Start

Basic Connection (Plain TCP)

# Connect to router
{:ok, conn} = RouterosApi.connect(%{
  host: "192.168.88.1",
  port: 8728,
  username: "admin",
  password: "password"
})

# Execute a command
{:ok, interfaces} = RouterosApi.command(conn, ["/interface/print"])

# Result is a list of maps
[
  %{
    "name" => "ether1",
    "type" => "ether",
    "disabled" => false,
    "running" => true
  },
  ...
]

# Disconnect when done
RouterosApi.disconnect(conn)

Secure Connection (TLS)

# Auto-detect TLS from port 8729
{:ok, conn} = RouterosApi.connect(%{
  host: "192.168.88.1",
  port: 8729,  # TLS port - auto-detected
  username: "admin",
  password: "password",
  ssl_opts: [verify: :verify_none]  # For self-signed certificates
})

# Or explicit TLS with certificate verification
{:ok, conn} = RouterosApi.connect(%{
  host: "192.168.88.1",
  port: 8729,
  username: "admin",
  password: "password",
  ssl: true,
  ssl_opts: [
    verify: :verify_peer,
    cacertfile: "/path/to/ca.pem"
  ]
})

Self-Signed Certificates

For lab/testing environments with self-signed certificates:

{:ok, conn} = RouterosApi.connect(%{
  host: "192.168.88.1",
  port: 8729,
  username: "admin",
  password: "password",
  ssl_opts: [
    verify: :verify_none  # Disables certificate verification
  ]
})

Note:verify: :verify_none should only be used in lab/testing environments. For production, use proper certificates and verify: :verify_peer.

Usage Examples

List IP Addresses

{:ok, addresses} = RouterosApi.command(conn, ["/ip/address/print"])

Add IP Address

{:ok, _} = RouterosApi.command(conn, [
  "/ip/address/add",
  "=address=192.168.88.2/24",
  "=interface=bridge"
])

Query with Filters

{:ok, [interface]} = RouterosApi.command(conn, [
  "/interface/print",
  "?name=ether1"
])

Error Handling

case RouterosApi.command(conn, ["/interface/print"]) do
  {:ok, data} ->
    IO.inspect(data)

  {:error, %RouterosApi.Error{type: :trap, message: msg}} ->
    IO.puts("RouterOS error: #{msg}")

  {:error, %RouterosApi.Error{type: :fatal}} ->
    IO.puts("Fatal error - connection lost")

  {:error, reason} ->
    IO.puts("Network error: #{inspect(reason)}")
end

Configuration

Connection Options

All connection options:

{:ok, conn} = RouterosApi.connect(%{
  host: "192.168.88.1",        # Required: Router hostname or IP
  port: 8728,                  # Optional: Port (default: 8728 for TCP, 8729 for TLS)
  username: "admin",           # Required: RouterOS username
  password: "password",        # Required: RouterOS password
  timeout: 5000,               # Optional: Connection timeout in ms (default: 5000)
  ssl: false,                  # Optional: Force TLS (auto-detected from port)
  ssl_opts: []                 # Optional: SSL options (e.g., verify: :verify_none)
})

Connection Pooling

For production use with multiple concurrent requests, use connection pooling:

# In your application.ex
def start(_type, _args) do
  children = [
    {RouterosApi.Pool, [
      name: :main_router,
      host: "192.168.88.1",
      port: 8729,                    # Optional: Use TLS
      username: "admin",
      password: "password",
      pool_size: 10,                 # Number of connections in pool
      ssl_opts: [verify: :verify_none]  # For self-signed certs
    ]}
  ]

  opts = [strategy: :one_for_one, name: MyApp.Supervisor]
  Supervisor.start_link(children, opts)
end

# In your code - use pool name instead of connection PID
{:ok, interfaces} = RouterosApi.command(:main_router, ["/interface/print"])
{:ok, resource} = RouterosApi.Helpers.get_system_resource(:main_router)

Benefits of pooling:

Telemetry

The library emits telemetry events for monitoring:

Connection Events:

Command Events:

Pool Events:

Example telemetry handler:

:telemetry.attach_many(
  "routeros-api-handler",
  [
    [:routeros_api, :command, :stop],
    [:routeros_api, :command, :exception]
  ],
  &MyApp.Telemetry.handle_event/4,
  nil
)

defmodule MyApp.Telemetry do
  require Logger

  def handle_event([:routeros_api, :command, :stop], measurements, metadata, _config) do
    if measurements.duration > 1_000_000_000 do
      Logger.warning("Slow RouterOS command: #{metadata.command} (#{measurements.duration}ns)")
    end
  end

  def handle_event([:routeros_api, :command, :exception], _measurements, metadata, _config) do
    Logger.error("RouterOS command failed: #{metadata.command} - #{inspect(metadata.reason)}")
  end
end

Helper Functions

The library provides convenient helper functions for common operations:

alias RouterosApi.Helpers

# List all interfaces
{:ok, interfaces} = Helpers.list_interfaces(conn)

# Get specific interface
{:ok, interface} = Helpers.get_interface(conn, "ether1")

# List IP addresses
{:ok, addresses} = Helpers.list_ip_addresses(conn)

# Add IP address
{:ok, _} = Helpers.add_ip_address(conn, "192.168.1.1/24", "ether1")

# Get system information
{:ok, resource} = Helpers.get_system_resource(conn)

# Get/set router identity
{:ok, identity} = Helpers.get_identity(conn)
{:ok, _} = Helpers.set_identity(conn, "MyRouter")

# List firewall rules
{:ok, rules} = Helpers.list_firewall_rules(conn)

# List DHCP leases
{:ok, leases} = Helpers.list_dhcp_leases(conn)

All helpers work with both direct connections and connection pools.

Streaming API (Live Monitoring)

For commands that produce continuous output (like /interface/monitor-traffic), use the streaming API:

# Start a dedicated streaming connection
{:ok, stream_conn} = RouterosApi.Stream.connect(%{
  host: "192.168.88.1",
  username: "admin",
  password: "password"
})

# Create a stream for interface traffic monitoring
{:ok, stream} = RouterosApi.Stream.monitor(stream_conn, [
  "/interface/monitor-traffic",
  "=interface=ether1"
])

# Consume the stream - each item is a data map
stream
|> Stream.take(10)  # Take 10 samples
|> Enum.each(fn data ->
  rx = data["rx-bits-per-second"]
  tx = data["tx-bits-per-second"]
  IO.puts("RX: #{rx} bps, TX: #{tx} bps")
end)

# Disconnect when done
RouterosApi.Stream.disconnect(stream_conn)

Time-Limited Monitoring

RouterOS supports duration-limited monitoring:

{:ok, stream} = RouterosApi.Stream.monitor(stream_conn, [
  "/interface/monitor-traffic",
  "=interface=ether1",
  "=duration=30s"  # Router stops after 30 seconds
])

# Collect all samples
samples = Enum.to_list(stream)

Stream Transformations

Streams work with all standard Elixir Stream functions:

{:ok, stream} = RouterosApi.Stream.monitor(stream_conn, [
  "/interface/monitor-traffic",
  "=interface=ether1"
])

# Calculate 5-sample moving average
stream
|> Stream.chunk_every(5)
|> Stream.map(fn samples ->
  avg = samples
    |> Enum.map(&String.to_integer(&1["rx-bits-per-second"]))
    |> Enum.sum()
    |> div(5)
  %{average_rx_bps: avg}
end)
|> Stream.take(10)
|> Enum.each(&IO.inspect/1)

Streaming Limitations

Stream Telemetry Events

Authentication

The library automatically handles authentication for different RouterOS versions:

RouterOS 7.x and 6.43+ (Plain Text)

Modern RouterOS versions use plain text authentication:

# The library automatically detects and uses plain text auth
{:ok, conn} = RouterosApi.connect(%{
  host: "192.168.88.1",
  username: "admin",
  password: "password"
})

RouterOS pre-6.43 (MD5 Challenge-Response)

Older RouterOS versions use MD5 challenge-response authentication. The library automatically falls back to this method if plain text authentication fails:

# Same code works - automatic fallback
{:ok, conn} = RouterosApi.connect(%{
  host: "192.168.88.1",
  username: "admin",
  password: "password"
})

Note: The authentication method is automatically detected and handled. You don't need to specify which method to use.

Troubleshooting

Connection Issues

"Connection refused"

"Authentication failed"

SSL/TLS Certificate Errors

Performance

Slow Commands

Connection Timeouts

Documentation

Full documentation is available at https://hexdocs.pm/routeros_api.

Testing

The library includes comprehensive test coverage:

# Run unit tests only
mix test

# Run all tests including integration tests
mix test --include integration --include ssl_integration

# Run with coverage
mix test --cover

# Run Dialyzer type checking
mix dialyzer

# Run code quality checks
mix credo --strict

Test Coverage:

Compatibility

Tested with:

Supported RouterOS versions:

Development

Quality Tools

The project uses several tools to maintain code quality:

Running Quality Checks

# Format code
mix format

# Check formatting
mix format --check-formatted

# Run all quality checks
mix test && mix dialyzer && mix credo

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

License

MIT License - see LICENSE for details.

Acknowledgments

This project was inspired by the original erotik Erlang library.