# snmpkit

High-performance SNMP toolkit for Python, powered by Rust.

## Installation

```bash
pip install snmpkit
```

Requires Python 3.11+.

## Quick Start

### SNMP Manager

Query SNMP devices:

```python
import asyncio
from snmpkit.manager import Manager

async def main():
    async with Manager("192.168.1.1", community="public") as mgr:
        # Get system description
        descr = await mgr.get("1.3.6.1.2.1.1.1.0")
        print(f"Device: {descr}")

        # Walk interface table
        async for oid, value in mgr.bulk_walk("1.3.6.1.2.1.2.2.1.2"):
            print(f"{oid} = {value}")

asyncio.run(main())
```

### AgentX Subagent

Extend an SNMP agent with custom data:

```python
from snmpkit.agent import Agent, Updater

class MyUpdater(Updater):
    async def update(self):
        self.set_INTEGER("1.0", 42)
        self.set_OCTETSTRING("2.0", "hello")

agent = Agent()
agent.register("1.3.6.1.4.1.12345", MyUpdater())

# Async (default)
await agent.start()

# Or sync
agent.start_sync()
```

## Manager API Reference

### Manager

```python
Manager(
    host: str,
    port: int = 161,
    # v1/v2c
    community: str = "public",
    version: int = 2,              # 1, 2 (v2c), or 3
    # v3 USM
    user: str | None = None,
    auth_protocol: str | None = None,  # MD5, SHA, SHA224, SHA256, SHA384, SHA512
    auth_password: str | None = None,
    priv_protocol: str | None = None,  # DES, AES128; AES192, AES256 (v1.8.0+)
    priv_password: str | None = None,
    context_name: str = "",
    # Common
    timeout: float = 5.0,
    retries: int = 3,
)
```

SNMPv3 usage:

```python
# authPriv
async with Manager(
    "192.168.1.1",
    version=3,
    user="admin",
    auth_protocol="SHA256",
    auth_password="authpass123",
    priv_protocol="AES128",
    priv_password="privpass123",
) as mgr:
    value = await mgr.get("1.3.6.1.2.1.1.1.0")

# authNoPriv
async with Manager(
    "192.168.1.1",
    version=3,
    user="readonly",
    auth_protocol="SHA",
    auth_password="authpass",
) as mgr:
    value = await mgr.get("1.3.6.1.2.1.1.1.0")

# noAuthNoPriv
async with Manager("192.168.1.1", version=3, user="noauth") as mgr:
    value = await mgr.get("1.3.6.1.2.1.1.1.0")
```

Engine discovery is automatic on connect for SNMPv3.

Methods:
- `get(oid: str) -> Value` — Get single OID
- `get_many(*oids: str, raise_exceptions: bool = True) -> list[Value]` — Get multiple OIDs. Pass `raise_exceptions=False` to return NoSuchObject/NoSuchInstance/EndOfMibView as values instead of raising.
- `get_next(oid: str) -> tuple[str, Value]` — Get next OID
- `get_bulk(*oids, non_repeaters=0, max_repetitions=10) -> list[tuple[str, Value]]` — Bulk get
- `set(oid: str, value: Value) -> None` — Set single OID value
- `set_many(*varbinds: tuple[str, Value]) -> None` — Set multiple OID/value pairs in one SET PDU
- `walk(oid: str) -> AsyncIterator[tuple[str, Value]]` — Walk using GetNext
- `bulk_walk(oid: str, bulk_size=10) -> AsyncIterator[tuple[str, Value]]` — Walk using GetBulk
- `get_table(base_oid, columns=None, bulk_size=25) -> dict[tuple, dict[int, Value]]` — Structured table
- `send_trap(trap_oid, varbinds=None, uptime=None) -> None` — Fire-and-forget trap
- `send_inform(trap_oid, varbinds=None, uptime=None) -> None` — Inform with ACK wait

### TrapReceiver

```python
from snmpkit.manager import TrapReceiver

async with TrapReceiver(port=1162) as receiver:
    async for trap in receiver:
        print(f"{trap.source}: {trap.trap_oid}")
        for oid, value in trap.varbinds:
            print(f"  {oid} = {value}")
```

TrapMessage fields: source, version, community, pdu_type, request_id, trap_oid, uptime, varbinds, raw_varbinds.
InformRequests are automatically acknowledged.

### Concurrent Polling

```python
from snmpkit.manager import PollTarget, poll_many

targets = [PollTarget(host="10.0.0.1"), PollTarget(host="10.0.0.2")]
async for result in poll_many(targets, ["1.3.6.1.2.1.1.1.0"], concurrency=50):
    print(f"{result.target}: {result.value or result.error}")
```

### Exceptions

```python
from snmpkit.manager.exceptions import (
    SnmpError,            # Base
    TimeoutError,         # Request timed out
    UnreachableError,     # DNS failure, refused, or no route
    AuthenticationError,  # SNMPv3 key or protocol mismatch
    NoSuchObjectError,    # OID doesn't exist
    NoSuchInstanceError,  # Instance doesn't exist
    EndOfMibViewError,    # End of MIB reached
    GenericError,         # SNMP error with status code
)
```

Every exception carries two flags (v1.7.0+). Branch on these rather than on the
exception class:

```python
except SnmpError as e:
    if e.unreachable:      # device never answered
        mark_offline(host)
    elif e.auth_failed:    # device answered, credentials rejected
        mark_credentials_invalid(host)
    else:                  # device answered, object is missing
        mark_object_missing(host)
```

## MIB API Reference (v1.7.0+)

Parses SMIv2 (RFC 2578/2579/2580) and SMIv1 (RFC 1155/1212/1215) into one OID tree.

### MibTree

```python
from snmpkit.mib import MibTree

tree = MibTree()
tree.load_dir("/usr/share/snmp/mibs")   # returns the number of modules parsed
tree.load_file("VENDOR-MIB")
tree.load_str(text)

node = tree.lookup("sysUpTime")          # name, MODULE::name, or numeric OID
node = tree.lookup("1.3.6.1.2.1.1.3")
tree.translate("1.3.6.1.2.1.2.2.1.2.3")  # 'IF-MIB::ifDescr.3'
tree.nearest("1.3.6.1.2.1.1.3.0")        # node, ignoring the instance suffix

tree.children("ifTable")
tree.walk("ifTable")                     # depth-first, OID order
tree.roots, tree.modules, tree.diagnostics
"ifDescr" in tree, len(tree), tree["ifDescr"]
```

### MibNode

```python
node.name, node.module, node.oid, node.numeric_oid
node.kind          # node | scalar | table | row | column | notification
node.syntax        # 'DisplayString' - as the MIB wrote it
node.base_type     # 'OCTET STRING'  - what it resolves to
node.max_access, node.status, node.description, node.units
node.display_hint, node.enums, node.defval
node.index, node.implied, node.augments, node.row_type, node.objects
node.parent, node.children, node.columns

node.format(1)                # 'up'   - enum label
node.format(b"\x00\x1a\x2b")  # '00:1a:2b' - DISPLAY-HINT, RFC 2579 3.1
node.enum_name(1), node.enum_value("up")
```

### Using MIB names with the Manager

```python
from snmpkit.manager import Manager

async with Manager("192.0.2.1", community="public", mib=tree) as mgr:
    await mgr.get("sysDescr.0")          # names work anywhere an OID does
    await mgr.get("IF-MIB::ifNumber.0")
    await mgr.get("1.3.6.1.2.1.1.1.0")   # numeric still works

    mgr.resolve("ifDescr.1")             # -> '1.3.6.1.2.1.2.2.1.2.1'
    mgr.translate("1.3.6.1.2.1.1.3.0")   # -> 'SNMPv2-MIB::sysUpTime.0'
    mgr.format("ifOperStatus.1", value)  # -> 'up'
```

## Agent API Reference

### Agent

```python
Agent(
    agent_id: str = "snmpkit",
    socket_path: str = "/var/agentx/master",
    parallel_encoding: bool = False,  # Enable Rust parallel encoding
    worker_threads: int = 0,          # Enable Python thread pool
    queue_size: int = 0,              # 0 = unbounded
)
```

Methods:
- `register(oid, updater, freq=10, context=None, priority=127)` — Register OID subtree
- `unregister(oid)` — Unregister OID subtree
- `start()` — Async coroutine
- `start_sync()` — Blocking
- `stop()` — Stop agent

### Updater

```python
class Updater:
    async def update(self) -> None: ...

    # Value setters
    def set_INTEGER(self, oid: str, value: int) -> None
    def set_OCTETSTRING(self, oid: str, value: str | bytes) -> None
    def set_OBJECTIDENTIFIER(self, oid: str, value: str) -> None
    def set_IPADDRESS(self, oid: str, value: str) -> None
    def set_COUNTER32(self, oid: str, value: int) -> None
    def set_GAUGE32(self, oid: str, value: int) -> None
    def set_TIMETICKS(self, oid: str, value: int) -> None
    def set_OPAQUE(self, oid: str, value: bytes) -> None
    def set_COUNTER64(self, oid: str, value: int) -> None
```

### SetHandler

```python
class SetHandler:
    async def test(self, oid: str, value) -> bool: ...
    async def commit(self, oid: str, value) -> None: ...
    async def undo(self, oid: str) -> None: ...
    async def cleanup(self, oid: str) -> None: ...
```

## Low-Level Rust Bindings

```python
from snmpkit.core import Oid, Value, VarBind

# OID manipulation
oid = Oid("1.3.6.1.4.1.12345")
print(oid.parts)        # [1, 3, 6, 1, 4, 1, 12345]
print(oid.parent())     # Oid("1.3.6.1.4.1")
print(oid.child(1))     # Oid("1.3.6.1.4.1.12345.1")

# SNMP values
val = Value.Integer(42)
val = Value.OctetString(b"hello")
val = Value.Counter64(9999999999)
val = Value.IpAddress(192, 168, 1, 1)

# Python conversions (v1.6.0+)
int(Value.Integer(42))         # 42
float(Value.Counter32(100))    # 100.0
bytes(Value.OctetString(b"x")) # b"x"
bool(Value.NoSuchObject())     # False
{Value.Integer(1)}             # hashable, usable in sets

# Variable bindings
vb = VarBind(oid, val)
```

## Performance

| Component | Operation | vs Competition |
|-----------|-----------|----------------|
| Manager | BER Encode | 127x faster than pysnmp |
| Manager | GET Request | 8.5x faster than pysnmp |
| Manager | WALK | 14x faster than pysnmp |
| Agent | PDU Encode | 11.5x faster than pyagentx3 |

## Entry Points

| Function | Type | Description |
|----------|------|-------------|
| `snmpkit.run(coro)` | Blocking | Entry point for async apps (wraps uvloop) |
| `agent.start()` | Async | Coroutine, use with `await` |
| `agent.start_sync()` | Blocking | For sync scripts |

Users never need to import uvloop directly.

## Links

- Documentation: https://snmpkit.dev
- GitHub: https://github.com/anthropics/snmpkit
- PyPI: https://pypi.org/project/snmpkit
