::=     定义为（产生式）
{}      集合或序列的界定符
[]      可选项或标签的界定符
()      子类型或值的界定符
|       选择符
..      范围符
--      注释符

-- 枚举类型示例
EthernetAdapterStatus ::= ENUMERATED {
    normal(0),
    degraded(1), 
    offline(2),
    failed(3)
}

-- 整数类型示例
EthernetNumberCollisionsRange ::= INTEGER {
    minimum(0),
    maximum(1000)
}

-- 对象标识符示例
internet OBJECT IDENTIFIER ::= { 
    iso(1) org(3) dod(6) 1 
}
-- 值为：1.3.6.1 或 iso.org.dod.1

-- 序列类型示例
Person ::= SEQUENCE {
    name    IA5String,
    age     INTEGER OPTIONAL,
    address IA5String DEFAULT "Unknown"
}

-- 序列值示例
person Person ::= {name "Tony", age 25}

-- 相同类型元素的序列
Seats ::= SEQUENCE OF INTEGER
seats Seats ::= {20, 30, 40}

-- 集合类型示例
Person ::= SET {
    age     INTEGER,
    name    IA5String
}
person Person ::= {25, "Tony"}  -- 顺序可变

VipSeats ::= SET OF INTEGER
vipseats VipSeats ::= {60, 80, 120}

-- 标签类型示例
Parentage ::= SET {
    subjectName [1] IMPLICIT IA5String,
    motherName  [2] IMPLICIT IA5String OPTIONAL,
    fatherName  [3] IMPLICIT IA5String OPTIONAL
}

-- CHOICE类型示例
Prize ::= CHOICE {
    car     IA5String,
    cash    INTEGER,
    nothing BOOLEAN
}

-- ANY类型示例
SoftwareVersion ::= ANY
-- 或
SoftwareVersion ::= ANY DEFINED BY INTEGER

SmallPrime ::= INTEGER (2|3|5|7|11|13|17|19|23|29)

First-half ::= Months (INCLUDES First-quarter | INCLUDES Second-quarter)

PositiveInteger ::= INTEGER (0<..PLUS-INFINITY)
NegativeInteger ::= INTEGER (MINUS-INFINITY..<0)

TouchToneButtons ::= IA5String (FROM("0"|"1"|"2"|"3"|"4"|"5"|"6"|"7"|"8"|"9"|"*"|"#"))
DigitString ::= IA5String (FROM("0"|"1"|"2"|"3"|"4"|"5"|"6"|"7"|"8"|"9"))

X25DataNumber ::= DigitString (SIZE(5..14))
ParameterList ::= SET SIZE(0..12) OF Parameter

TestPDU ::= PDU (WITH COMPONENTS {
    ...,
    delta(FALSE),
    alpha(MIN..<0)
})

+----------+----------+----------+
| Type Tag | Length   | Value    |
+----------+----------+----------+

+---+---+---+---+---+---+---+---+
| 7 | 6 | 5 | 4 | 3 | 2 | 1 | 0 |
+---+---+---+---+---+---+---+---+
|  类型  |构造|    标签值        |
|  (2位) |标志|    (5位)         |
+-------+----+------------------+

FALSE: 01 01 00
TRUE:  01 01 FF

十进制256: 02 02 01 00

比特串10101: 03 02 03 A8

SEQUENCE{madeofwood TRUE, length 62}:
30 06 01 01 FF 02 01 3E

MACRO-NAME MACRO ::= BEGIN
    TYPE NOTATION ::= type-notation
    VALUE NOTATION ::= value-notation
    supporting-productions
END

OBJECT-TYPE MACRO ::= BEGIN
    TYPE NOTATION ::= 
        "SYNTAX" type(Syntax)
        "ACCESS" Access
        "STATUS" Status
        "DESCRIPTION" Text
        "INDEX" "{" IndexTypes "}"
        "DEFVAL" "{" Defval "}"
    VALUE NOTATION ::= value(VALUE ObjectName)
    Access ::= "read-only" | "read-write" | "write-only" | "not-accessible"
    Status ::= "mandatory" | "optional" | "obsolete" | "deprecated"
END

-- 个人记录定义
PersonnelRecord ::= [APPLICATION 0] IMPLICIT SET {
    name            Name,
    title           [0] VisibleString,
    number          EmployeeNumber,
    dateOfHire      [1] Date,
    nameOfSpouse    [2] Name,
    children        [3] IMPLICIT SEQUENCE OF ChildInformation DEFAULT {}
}

Name ::= [APPLICATION 1] IMPLICIT SEQUENCE {
    givenName       VisibleString,
    initial         VisibleString,
    familyName      VisibleString
}

EmployeeNumber ::= [APPLICATION 2] IMPLICIT INTEGER

Date ::= [APPLICATION 3] IMPLICIT VisibleString -- YYYYMMDD

ChildInformation ::= SET {
    name            Name,
    dateOfBirth     [0] Date
}

-- 以太网接口对象
EthernetInterface ::= SEQUENCE {
    ifIndex         INTEGER,
    ifDescr         DisplayString,
    ifType          INTEGER,
    ifMtu           INTEGER,
    ifSpeed         Gauge,
    ifPhysAddress   PhysAddress,
    ifAdminStatus   INTEGER {
                        up(1),
                        down(2),
                        testing(3)
                    },
    ifOperStatus    INTEGER {
                        up(1),
                        down(2),
                        testing(3)
                    }
}

pip install pyasn1

from pyasn1.type import univ, namedtype, tag
from pyasn1.codec.der import encoder, decoder

# 定义简单类型
boolean_val = univ.Boolean(True)
integer_val = univ.Integer(256)
octet_string_val = univ.OctetString(b'Hello')

# 编码
encoded_bool = encoder.encode(boolean_val)
encoded_int = encoder.encode(integer_val)
encoded_str = encoder.encode(octet_string_val)

print(f"Boolean编码: {encoded_bool.hex()}")
print(f"Integer编码: {encoded_int.hex()}")
print(f"OctetString编码: {encoded_str.hex()}")

# 解码
decoded_bool, _ = decoder.decode(encoded_bool, asn1Spec=univ.Boolean())
decoded_int, _ = decoder.decode(encoded_int, asn1Spec=univ.Integer())
decoded_str, _ = decoder.decode(encoded_str, asn1Spec=univ.OctetString())

print(f"Boolean解码: {decoded_bool}")
print(f"Integer解码: {decoded_int}")
print(f"OctetString解码: {decoded_str}")

from pyasn1.type import univ, namedtype

# 定义Person序列类型
class Person(univ.Sequence):
    componentType = namedtype.NamedTypes(
        namedtype.NamedType('name', univ.UTF8String()),
        namedtype.NamedType('age', univ.Integer())
    )

# 创建Person实例
person = Person()
person.setComponentByName('name', 'Tony')
person.setComponentByName('age', 25)

# 编码
encoded_person = encoder.encode(person)
print(f"Person编码: {encoded_person.hex()}")

# 解码
decoded_person, _ = decoder.decode(encoded_person, asn1Spec=Person())
print(f"姓名: {decoded_person.getComponentByName('name')}")
print(f"年龄: {decoded_person.getComponentByName('age')}")

from pyasn1.type import univ, namedtype, char

# 定义Prize CHOICE类型
class Prize(univ.Choice):
    componentType = namedtype.NamedTypes(
        namedtype.NamedType('car', char.UTF8String()),
        namedtype.NamedType('cash', univ.Integer()),
        namedtype.NamedType('nothing', univ.Boolean())
    )

# 创建不同的Prize实例
car_prize = Prize()
car_prize.setComponentByName('car', 'BMW')

cash_prize = Prize()
cash_prize.setComponentByName('cash', 10000)

# 编码
encoded_car = encoder.encode(car_prize)
encoded_cash = encoder.encode(cash_prize)

print(f"Car prize编码: {encoded_car.hex()}")
print(f"Cash prize编码: {encoded_cash.hex()}")

# 安装asn1c
sudo apt-get install asn1c

# 编译ASN.1文件
asn1c -fcompound-names PersonnelRecord.asn1

# 生成的C代码文件
ls *.c *.h

[[include]] "PersonnelRecord.h"
[[include]] <stdio.h>

int main() {
    PersonnelRecord_t *record;
    Name_t *name;
    
    // 创建记录
    record = calloc(1, sizeof(PersonnelRecord_t));
    name = calloc(1, sizeof(Name_t));
    
    // 设置姓名
    OCTET_STRING_fromString(&name->givenName, "John");
    OCTET_STRING_fromString(&name->initial, "D");
    OCTET_STRING_fromString(&name->familyName, "Smith");
    
    record->name = *name;
    
    // 设置员工编号
    record->number = 12345;
    
    // 编码
    asn_enc_rval_t ec;
    ec = der_encode_to_buffer(&asn_DEF_PersonnelRecord, record, buffer, sizeof(buffer));
    
    if(ec.encoded == -1) {
        printf("编码失败\n");
    } else {
        printf("编码成功，长度: %ld\n", ec.encoded);
    }
    
    return 0;
}

+------------------+
|   管理应用程序    |
+------------------+
|      SNMP        |
+------------------+
|       UDP        |
+------------------+
|       IP         |
+------------------+
|   网络访问层      |
+------------------+

NetworkAddress ::= CHOICE {
    internet IpAddress
}

IpAddress ::= [APPLICATION 0] IMPLICIT OCTET STRING (SIZE(4))

Counter ::= [APPLICATION 1] IMPLICIT INTEGER (0..4294967295)

Gauge ::= [APPLICATION 2] IMPLICIT INTEGER (0..4294967295)

TimeTicks ::= [APPLICATION 3] IMPLICIT INTEGER (0..4294967295)

Opaque ::= [APPLICATION 4] IMPLICIT OCTET STRING

OBJECT-TYPE MACRO ::= BEGIN
    TYPE NOTATION ::= 
        "SYNTAX" type(TYPE ObjectSyntax)
        "ACCESS" Access
        "STATUS" Status
        DescrPart
        ReferPart
        IndexPart
        DefValPart
    VALUE NOTATION ::= value(VALUE ObjectName)
    
    Access ::= "read-only" | "read-write" | "write-only" | "not-accessible"
    Status ::= "mandatory" | "optional" | "obsolete" | "deprecated"
    
    DescrPart ::= "DESCRIPTION" value(description DisplayString) | empty
    ReferPart ::= "REFERENCE" value(reference DisplayString) | empty
    IndexPart ::= "INDEX" "{" IndexTypes "}"
    DefValPart ::= "DEFVAL" "{" value(defvalue ObjectSyntax) "}" | empty
END

tcpMaxConn OBJECT-TYPE
    SYNTAX      INTEGER
    ACCESS      read-only
    STATUS      mandatory
    DESCRIPTION "The limit on the total number of TCP connections 
                 the entity can support."
    ::= { tcp 4 }

sysDescr OBJECT-TYPE
    SYNTAX      DisplayString (SIZE (0..255))
    ACCESS      read-only
    STATUS      mandatory
    DESCRIPTION "A textual description of the entity."
    ::= { system 1 }

-- TCP连接表
tcpConnTable OBJECT-TYPE
    SYNTAX      SEQUENCE OF TcpConnEntry
    ACCESS      not-accessible
    STATUS      mandatory
    DESCRIPTION "A table containing TCP connection-specific information."
    ::= { tcp 13 }

-- TCP连接表条目
tcpConnEntry OBJECT-TYPE
    SYNTAX      TcpConnEntry
    ACCESS      not-accessible
    STATUS      mandatory
    DESCRIPTION "Information about a particular current TCP connection."
    INDEX       { tcpConnLocalAddress, tcpConnLocalPort,
                  tcpConnRemAddress, tcpConnRemPort }
    ::= { tcpConnTable 1 }

-- TCP连接表条目结构
TcpConnEntry ::= SEQUENCE {
    tcpConnState        INTEGER,
    tcpConnLocalAddress IpAddress,
    tcpConnLocalPort    INTEGER (0..65535),
    tcpConnRemAddress   IpAddress,
    tcpConnRemPort      INTEGER (0..65535)
}

-- 连接状态
tcpConnState OBJECT-TYPE
    SYNTAX      INTEGER {
                    closed(1),
                    listen(2),
                    synSent(3),
                    synReceived(4),
                    established(5),
                    finWait1(6),
                    finWait2(7),
                    closeWait(8),
                    lastAck(9),
                    closing(10),
                    timeWait(11)
                }
    ACCESS      read-write
    STATUS      mandatory
    DESCRIPTION "The state of this TCP connection."
    ::= { tcpConnEntry 1 }

-- 系统描述
sysDescr OBJECT-TYPE
    SYNTAX      DisplayString (SIZE (0..255))
    ACCESS      read-only
    STATUS      mandatory
    ::= { system 1 }

-- 系统对象标识符
sysObjectID OBJECT-TYPE
    SYNTAX      OBJECT IDENTIFIER
    ACCESS      read-only
    STATUS      mandatory
    ::= { system 2 }

-- 系统启动时间
sysUpTime OBJECT-TYPE
    SYNTAX      TimeTicks
    ACCESS      read-only
    STATUS      mandatory
    ::= { system 3 }

-- 系统联系人
sysContact OBJECT-TYPE
    SYNTAX      DisplayString (SIZE (0..255))
    ACCESS      read-write
    STATUS      mandatory
    ::= { system 4 }

-- 系统名称
sysName OBJECT-TYPE
    SYNTAX      DisplayString (SIZE (0..255))
    ACCESS      read-write
    STATUS      mandatory
    ::= { system 5 }

-- 系统位置
sysLocation OBJECT-TYPE
    SYNTAX      DisplayString (SIZE (0..255))
    ACCESS      read-write
    STATUS      mandatory
    ::= { system 6 }

-- 系统服务
sysServices OBJECT-TYPE
    SYNTAX      INTEGER (0..127)
    ACCESS      read-only
    STATUS      mandatory
    ::= { system 7 }

-- 接口数量
ifNumber OBJECT-TYPE
    SYNTAX      INTEGER
    ACCESS      read-only
    STATUS      mandatory
    DESCRIPTION "The number of network interfaces present on this system."
    ::= { interfaces 1 }

-- 接口表
ifTable OBJECT-TYPE
    SYNTAX      SEQUENCE OF IfEntry
    ACCESS      not-accessible
    STATUS      mandatory
    ::= { interfaces 2 }

ifEntry OBJECT-TYPE
    SYNTAX      IfEntry
    ACCESS      not-accessible
    STATUS      mandatory
    INDEX       { ifIndex }
    ::= { ifTable 1 }

IfEntry ::= SEQUENCE {
    ifIndex             INTEGER,
    ifDescr             DisplayString,
    ifType              INTEGER,
    ifMtu               INTEGER,
    ifSpeed             Gauge,
    ifPhysAddress       PhysAddress,
    ifAdminStatus       INTEGER,
    ifOperStatus        INTEGER,
    ifLastChange        TimeTicks,
    ifInOctets          Counter,
    ifInUcastPkts       Counter,
    ifInNUcastPkts      Counter,
    ifInDiscards        Counter,
    ifInErrors          Counter,
    ifInUnknownProtos   Counter,
    ifOutOctets         Counter,
    ifOutUcastPkts      Counter,
    ifOutNUcastPkts     Counter,
    ifOutDiscards       Counter,
    ifOutErrors         Counter,
    ifOutQLen           Gauge,
    ifSpecific          OBJECT IDENTIFIER
}

-- IP转发开关
ipForwarding OBJECT-TYPE
    SYNTAX      INTEGER {
                    forwarding(1),
                    not-forwarding(2)
                }
    ACCESS      read-write
    STATUS      mandatory
    ::= { ip 1 }

-- IP默认TTL
ipDefaultTTL OBJECT-TYPE
    SYNTAX      INTEGER (1..255)
    ACCESS      read-write
    STATUS      mandatory
    ::= { ip 2 }

-- IP统计信息
ipInReceives OBJECT-TYPE
    SYNTAX      Counter
    ACCESS      read-only
    STATUS      mandatory
    ::= { ip 3 }

-- IP地址表
ipAddrTable OBJECT-TYPE
    SYNTAX      SEQUENCE OF IpAddrEntry
    ACCESS      not-accessible
    STATUS      mandatory
    ::= { ip 20 }

-- IP路由表
ipRouteTable OBJECT-TYPE
    SYNTAX      SEQUENCE OF IpRouteEntry
    ACCESS      not-accessible
    STATUS      mandatory
    ::= { ip 21 }

pip install pysnmp

from pysnmp.hlapi import *

def snmp_get(target, oid, community='public'):
    """SNMP GET操作"""
    for (errorIndication, errorStatus, errorIndex, varBinds) in nextCmd(
        SnmpEngine(),
        CommunityData(community),
        UdpTransportTarget((target, 161)),
        ContextData(),
        ObjectType(ObjectIdentity(oid)),
        lexicographicMode=False):
        
        if errorIndication:
            print(f"错误: {errorIndication}")
            break
        elif errorStatus:
            print(f"错误: {errorStatus.prettyPrint()} at {errorIndex and varBinds[int(errorIndex) - 1][0] or '?'}")
            break
        else:
            for varBind in varBinds:
                print(f"{varBind[0]} = {varBind[1]}")

def snmp_walk(target, oid, community='public'):
    """SNMP WALK操作"""
    for (errorIndication, errorStatus, errorIndex, varBinds) in nextCmd(
        SnmpEngine(),
        CommunityData(community),
        UdpTransportTarget((target, 161)),
        ContextData(),
        ObjectType(ObjectIdentity(oid)),
        lexicographicMode=False,
        ignoreNonIncreasingOid=False):
        
        if errorIndication:
            print(f"错误: {errorIndication}")
            break
        elif errorStatus:
            print(f"错误: {errorStatus.prettyPrint()}")
            break
        else:
            for varBind in varBinds:
                print(f"{varBind[0]} = {varBind[1]}")

# 使用示例
if __name__ == "__main__":
    target_host = "192.168.1.1"
    
    # 获取系统描述
    print("系统描述:")
    snmp_get(target_host, "1.3.6.1.2.1.1.1.0")
    
    # 获取系统启动时间
    print("\n系统启动时间:")
    snmp_get(target_host, "1.3.6.1.2.1.1.3.0")
    
    # 遍历接口表
    print("\n接口信息:")
    snmp_walk(target_host, "1.3.6.1.2.1.2.2.1.2")

from pysnmp.smi import builder, view, compiler

def load_mib_module():
    """加载MIB模块"""
    # 创建MIB构建器
    mibBuilder = builder.MibBuilder()
    
    # 加载标准MIB模块
    mibBuilder.loadModules('SNMPv2-MIB', 'IF-MIB', 'IP-MIB')
    
    # 创建MIB视图
    mibView = view.MibViewController(mibBuilder)
    
    return mibView

def resolve_oid(mibView, oid_name):
    """解析OID名称到数字OID"""
    try:
        oid, label, suffix = mibView.getNodeByName(oid_name)
        return '.'.join([str(x) for x in oid])
    except Exception as e:
        print(f"解析OID失败: {e}")
        return None

def resolve_name(mibView, oid):
    """解析数字OID到名称"""
    try:
        oid_tuple = tuple([int(x) for x in oid.split('.')])
        name, label, suffix = mibView.getNodeByOid(oid_tuple)
        return str(name)
    except Exception as e:
        print(f"解析名称失败: {e}")
        return None

# 使用示例
mibView = load_mib_module()

# OID名称到数字
print("sysDescr OID:", resolve_oid(mibView, ('SNMPv2-MIB', 'sysDescr')))
print("ifNumber OID:", resolve_oid(mibView, ('IF-MIB', 'ifNumber')))

# 数字OID到名称
print("1.3.6.1.2.1.1.1 名称:", resolve_name(mibView, "1.3.6.1.2.1.1.1"))

EXAMPLE-MIB DEFINITIONS ::= BEGIN

IMPORTS
    MODULE-IDENTITY, OBJECT-TYPE, Integer32, enterprises
        FROM SNMPv2-SMI
    DisplayString
        FROM SNMPv2-TC;

exampleMIB MODULE-IDENTITY
    LAST-UPDATED "202406080000Z"
    ORGANIZATION "Example Organization"
    CONTACT-INFO "contact@example.com"
    DESCRIPTION  "Example MIB module"
    ::= { enterprises 12345 }

-- 定义自定义对象
exampleObjects OBJECT IDENTIFIER ::= { exampleMIB 1 }

-- 服务器温度
serverTemperature OBJECT-TYPE
    SYNTAX      Integer32 (0..100)
    UNITS       "degrees Celsius"
    MAX-ACCESS  read-only
    STATUS      current
    DESCRIPTION "Current server temperature in degrees Celsius"
    ::= { exampleObjects 1 }

-- 服务器状态
serverStatus OBJECT-TYPE
    SYNTAX      INTEGER {
                    running(1),
                    stopped(2),
                    maintenance(3),
                    error(4)
                }
    MAX-ACCESS  read-write
    STATUS      current
    DESCRIPTION "Current server status"
    ::= { exampleObjects 2 }

-- 服务器描述
serverDescription OBJECT-TYPE
    SYNTAX      DisplayString (SIZE(0..255))
    MAX-ACCESS  read-write
    STATUS      current
    DESCRIPTION "Server description"
    ::= { exampleObjects 3 }

END

+-------------+-------------+------------------+
| 版本号      | 团体名      | 协议数据单元(PDU) |
+-------------+-------------+------------------+

def process_get_request(request):
    """处理Get请求的伪代码"""
    response = create_response(request.request_id)
    
    for variable in request.variable_bindings:
        try:
            # 检查对象是否存在
            if not object_exists(variable.name):
                return error_response("noSuchName", variable.index)
            
            # 检查访问权限
            if not has_read_access(variable.name):
                return error_response("noSuchName", variable.index)
            
            # 获取对象值
            value = get_object_value(variable.name)
            response.add_variable(variable.name, value)
            
        except TooBigError:
            return error_response("tooBig", 0)
        except Exception:
            return error_response("genError", variable.index)
    
    return response

# 请求示例
get_request = GetRequest(
    request_id=12345,
    variable_bindings=[
        ("1.3.6.1.2.1.7.1.0", None),  # udpInDatagrams
        ("1.3.6.1.2.1.7.2.0", None),  # udpNoPorts
        ("1.3.6.1.2.1.7.3.0", None),  # udpInErrors
        ("1.3.6.1.2.1.7.4.0", None)   # udpOutDatagrams
    ]
)

# 响应示例
get_response = GetResponse(
    request_id=12345,
    error_status=0,
    error_index=0,
    variable_bindings=[
        ("1.3.6.1.2.1.7.1.0", 100),
        ("1.3.6.1.2.1.7.2.0", 1),
        ("1.3.6.1.2.1.7.3.0", 2),
        ("1.3.6.1.2.1.7.4.0", 200)
    ]
)

def process_getnext_request(request):
    """处理GetNext请求的伪代码"""
    response = create_response(request.request_id)
    
    for variable in request.variable_bindings:
        try:
            # 查找下一个对象
            next_oid = find_next_object(variable.name)
            
            if next_oid is None:
                return error_response("noSuchName", variable.index)
            
            # 检查访问权限
            if not has_read_access(next_oid):
                # 继续查找下一个可访问对象
                next_oid = find_next_accessible_object(next_oid)
            
            if next_oid is None:
                return error_response("noSuchName", variable.index)
            
            value = get_object_value(next_oid)
            response.add_variable(next_oid, value)
            
        except Exception:
            return error_response("genError", variable.index)
    
    return response

# 使用GetNext发现MIB结构
def walk_mib_tree(agent, start_oid):
    """遍历MIB树"""
    current_oid = start_oid
    results = []
    
    while True:
        response = agent.get_next(current_oid)
        
        if response.error_status != 0:
            break
        
        next_oid, value = response.variable_bindings[0]
        
        # 检查是否还在目标子树中
        if not next_oid.startswith(start_oid):
            break
        
        results.append((next_oid, value))
        current_oid = next_oid
    
    return results

# 遍历UDP组
udp_objects = walk_mib_tree(agent, "1.3.6.1.2.1.7")

def walk_table(agent, table_oid):
    """遍历SNMP表"""
    table_data = []
    current_oid = table_oid
    
    while True:
        response = agent.get_next(current_oid)
        
        if response.error_status != 0:
            break
        
        next_oid, value = response.variable_bindings[0]
        
        # 检查是否还在表中
        if not next_oid.startswith(table_oid):
            break
        
        # 解析行索引
        row_index = extract_row_index(next_oid, table_oid)
        column_id = extract_column_id(next_oid, table_oid)
        
        table_data.append({
            'row_index': row_index,
            'column_id': column_id,
            'value': value
        })
        
        current_oid = next_oid
    
    return organize_table_data(table_data)

# 遍历接口表
interface_table = walk_table(agent, "1.3.6.1.2.1.2.2.1")

def process_set_request(request):
    """处理Set请求的伪代码"""
    response = create_response(request.request_id)
    
    # 第一阶段：验证所有变量
    for variable in request.variable_bindings:
        if not object_exists(variable.name):
            return error_response("noSuchName", variable.index)
        
        if not has_write_access(variable.name):
            return error_response("readOnly", variable.index)
        
        if not validate_value(variable.name, variable.value):
            return error_response("badValue", variable.index)
    
    # 第二阶段：执行所有设置操作
    try:
        for variable in request.variable_bindings:
            set_object_value(variable.name, variable.value)
            response.add_variable(variable.name, variable.value)
    except Exception:
        # 回滚所有更改
        rollback_changes()
        return error_response("genError", 0)
    
    return response

# 设置系统联系人信息
set_request = SetRequest(
    request_id=12346,
    variable_bindings=[
        ("1.3.6.1.2.1.1.4.0", "admin@company.com"),  # sysContact
        ("1.3.6.1.2.1.1.6.0", "Server Room A")       # sysLocation
    ]
)

# 设置接口管理状态
set_interface_status = SetRequest(
    request_id=12347,
    variable_bindings=[
        ("1.3.6.1.2.1.2.2.1.7.1", 1)  # ifAdminStatus.1 = up(1)
    ]
)

def generate_trap(trap_type, specific_code=0, variables=None):
    """生成SNMP Trap"""
    trap = TrapPDU(
        enterprise=get_enterprise_oid(),
        agent_addr=get_local_ip(),
        generic_trap=trap_type,
        specific_trap=specific_code,
        time_stamp=get_system_uptime(),
        variable_bindings=variables or []
    )
    
    return trap

# 链路断开陷阱
link_down_trap = generate_trap(
    trap_type=2,  # linkDown
    variables=[
        ("1.3.6.1.2.1.2.2.1.1.1", 1),  # ifIndex.1
        ("1.3.6.1.2.1.2.2.1.7.1", 2)   # ifAdminStatus.1 = down(2)
    ]
)

# 认证失败陷阱
auth_failure_trap = generate_trap(
    trap_type=4,  # authenticationFailure
    variables=[
        ("1.3.6.1.2.1.1.3.0", get_system_uptime())
    ]
)

def bulk_get(agent, oid_list, max_repetitions=10):
    """批量获取操作（模拟GetBulk）"""
    results = {}
    
    # 将OID列表分组，避免PDU过大
    for oid_group in chunk_list(oid_list, max_repetitions):
        request = GetRequest(
            request_id=generate_request_id(),
            variable_bindings=[(oid, None) for oid in oid_group]
        )
        
        response = agent.send_request(request)
        
        if response.error_status == 0:
            for oid, value in response.variable_bindings:
                results[oid] = value
        else:
            # 处理错误，可能需要单独重试
            handle_bulk_error(response, oid_group)
    
    return results

class SNMPCache:
    """SNMP响应缓存"""
    def __init__(self, ttl=60):
        self.cache = {}
        self.ttl = ttl
    
    def get(self, oid):
        """获取缓存值"""
        if oid in self.cache:
            value, timestamp = self.cache[oid]
            if time.time() - timestamp < self.ttl:
                return value
            else:
                del self.cache[oid]
        return None
    
    def set(self, oid, value):
        """设置缓存值"""
        self.cache[oid] = (value, time.time())
    
    def invalidate(self, oid_pattern):
        """失效匹配模式的缓存"""
        to_remove = [oid for oid in self.cache.keys() 
                    if oid.startswith(oid_pattern)]
        for oid in to_remove:
            del self.cache[oid]

def reliable_snmp_get(agent, oid, max_retries=3, timeout=5):
    """可靠的SNMP GET操作"""
    for attempt in range(max_retries):
        try:
            response = agent.get(oid, timeout=timeout)
            
            if response.error_status == 0:
                return response.variable_bindings[0][1]
            elif response.error_status == 2:  # noSuchName
                raise SNMPObjectNotFound(f"Object {oid} not found")
            else:
                raise SNMPError(f"SNMP error: {response.error_status}")
        
        except TimeoutError:
            if attempt == max_retries - 1:
                raise SNMPTimeout(f"Timeout after {max_retries} attempts")
            time.sleep(2 ** attempt)  # 指数退避
        
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(1)

class SNMPErrorHandler:
    """SNMP错误处理器"""
    
    @staticmethod
    def handle_error(error_status, error_index, variables):
        """处理SNMP错误"""
        if error_status == 1:  # tooBig
            return SNMPErrorHandler.handle_too_big(variables)
        elif error_status == 2:  # noSuchName
            return SNMPErrorHandler.handle_no_such_name(error_index, variables)
        elif error_status == 3:  # badValue
            return SNMPErrorHandler.handle_bad_value(error_index, variables)
        elif error_status == 4:  # readOnly
            return SNMPErrorHandler.handle_read_only(error_index, variables)
        elif error_status == 5:  # genError
            return SNMPErrorHandler.handle_general_error(error_index, variables)
    
    @staticmethod
    def handle_too_big(variables):
        """处理PDU过大错误"""
        # 分割变量列表，分批处理
        chunk_size = len(variables) // 2
        return [variables[:chunk_size], variables[chunk_size:]]
    
    @staticmethod
    def handle_no_such_name(error_index, variables):
        """处理对象不存在错误"""
        problematic_var = variables[error_index - 1]
        print(f"Object not found: {problematic_var[0]}")
        # 从列表中移除问题变量，重试其他变量
        return [var for i, var in enumerate(variables) if i != error_index - 1]

import socket
import time
from pyasn1.codec.der import encoder, decoder
from pyasn1.type import univ, namedtype, tag

class SNMPClient:
    """SNMP客户端实现"""
    
    def __init__(self, host, port=161, community="public", timeout=5, retries=3):
        self.host = host
        self.port = port
        self.community = community
        self.timeout = timeout
        self.retries = retries
        self.request_id = 1
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.socket.settimeout(timeout)
    
    def _next_request_id(self):
        """生成下一个请求ID"""
        self.request_id += 1
        return self.request_id
    
    def _create_pdu(self, pdu_type, variables):
        """创建PDU"""
        from pysnmp.proto import rfc1157
        
        pdu = rfc1157.GetRequestPDU()
        pdu.setComponentByName('request-id', self._next_request_id())
        pdu.setComponentByName('error-status', 0)
        pdu.setComponentByName('error-index', 0)
        
        var_binds = pdu.setComponentByName('variable-bindings')
        for i, (oid, value) in enumerate(variables):
            var_bind = var_binds.setComponentByPosition(i)
            var_bind.setComponentByName('name', oid)
            if value is not None:
                var_bind.setComponentByName('value', value)
        
        return pdu
    
    def _send_request(self, pdu):
        """发送SNMP请求"""
        # 创建SNMP消息
        message = self._create_message(pdu)
        
        # 编码并发送
        encoded_message = encoder.encode(message)
        
        for attempt in range(self.retries):
            try:
                self.socket.sendto(encoded_message, (self.host, self.port))
                response_data, addr = self.socket.recvfrom(65535)
                
                # 解码响应
                response_message, _ = decoder.decode(response_data)
                return self._parse_response(response_message)
                
            except socket.timeout:
                if attempt == self.retries - 1:
                    raise TimeoutError(f"SNMP timeout after {self.retries} attempts")
                continue
    
    def get(self, oid):
        """SNMP GET操作"""
        variables = [(oid, None)]
        pdu = self._create_pdu('get', variables)
        response = self._send_request(pdu)
        
        if response['error_status'] == 0:
            return response['variables'][0][1]
        else:
            raise SNMPError(f"SNMP error: {response['error_status']}")
    
    def get_next(self, oid):
        """SNMP GETNEXT操作"""
        variables = [(oid, None)]
        pdu = self._create_pdu('getnext', variables)
        response = self._send_request(pdu)
        
        if response['error_status'] == 0:
            return response['variables'][0]
        else:
            raise SNMPError(f"SNMP error: {response['error_status']}")
    
    def set(self, oid, value):
        """SNMP SET操作"""
        variables = [(oid, value)]
        pdu = self._create_pdu('set', variables)
        response = self._send_request(pdu)
        
        if response['error_status'] != 0:
            raise SNMPError(f"SNMP error: {response['error_status']}")
    
    def walk(self, start_oid):
        """SNMP WALK操作"""
        results = []
        current_oid = start_oid
        
        while True:
            try:
                next_oid, value = self.get_next(current_oid)
                
                # 检查是否还在目标子树中
                if not next_oid.startswith(start_oid):
                    break
                
                results.append((next_oid, value))
                current_oid = next_oid
                
            except SNMPError:
                break
        
        return results
    
    def close(self):
        """关闭连接"""
        self.socket.close()

# 使用示例
def main():
    client = SNMPClient("192.168.1.1", community="public")
    
    try:
        # 获取系统描述
        sys_descr = client.get("1.3.6.1.2.1.1.1.0")
        print(f"系统描述: {sys_descr}")
        
        # 获取系统启动时间
        sys_uptime = client.get("1.3.6.1.2.1.1.3.0")
        print(f"系统启动时间: {sys_uptime}")
        
        # 遍历接口表
        print("\n接口信息:")
        interfaces = client.walk("1.3.6.1.2.1.2.2.1.2")
        for oid, desc in interfaces:
            print(f"{oid}: {desc}")
        
        # 设置系统联系人
        client.set("1.3.6.1.2.1.1.4.0", "admin@example.com")
        print("系统联系人已更新")
        
    except Exception as e:
        print(f"错误: {e}")
    finally:
        client.close()

if __name__ == "__main__":
    main()

class SNMPAgent:
    """简单的SNMP代理实现"""
    
    def __init__(self, host="0.0.0.0", port=161, community="public"):
        self.host = host
        self.port = port
        self.community = community
        self.mib = {}
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.socket.bind((host, port))
        self.running = False
        
        # 初始化基本MIB对象
        self._init_system_mib()
    
    def _init_system_mib(self):
        """初始化系统MIB对象"""
        self.mib.update({
            "1.3.6.1.2.1.1.1.0": "Linux Router 1.0",  # sysDescr
            "1.3.6.1.2.1.1.2.0": "1.3.6.1.4.1.8072.3.2.10",  # sysObjectID
            "1.3.6.1.2.1.1.3.0": self._get_uptime,  # sysUpTime (函数)
            "1.3.6.1.2.1.1.4.0": "admin@localhost",  # sysContact
            "1.3.6.1.2.1.1.5.0": "localhost",  # sysName
            "1.3.6.1.2.1.1.6.0": "Unknown",  # sysLocation
            "1.3.6.1.2.1.1.7.0": 72,  # sysServices
        })
    
    def _get_uptime(self):
        """获取系统启动时间"""
        with open('/proc/uptime', 'r') as f:
            uptime_seconds = float(f.readline().split()[0])
        return int(uptime_seconds * 100)  # 转换为centiseconds
    
    def register_object(self, oid, value_or_function, access="read-only"):
        """注册MIB对象"""
        self.mib[oid] = {
            'value': value_or_function,
            'access': access
        }
    
    def get_object_value(self, oid):
        """获取对象值"""
        if oid in self.mib:
            obj = self.mib[oid]
            if callable(obj):
                return obj()
            elif isinstance(obj, dict):
                value = obj['value']
                return value() if callable(value) else value
            else:
                return obj
        else:
            raise SNMPError("noSuchName")
    
    def set_object_value(self, oid, value):
        """设置对象值"""
        if oid not in self.mib:
            raise SNMPError("noSuchName")
        
        obj = self.mib[oid]
        if isinstance(obj, dict):
            if obj['access'] == 'read-only':
                raise SNMPError("readOnly")
            obj['value'] = value
        else:
            raise SNMPError("readOnly")
    
    def find_next_object(self, oid):
        """查找下一个对象"""
        # 按词典顺序排序所有OID
        sorted_oids = sorted(self.mib.keys())
        
        for next_oid in sorted_oids:
            if next_oid > oid:
                return next_oid
        
        return None
    
    def process_get_request(self, pdu):
        """处理GET请求"""
        response_vars = []
        
        for var_oid, _ in pdu['variables']:
            try:
                value = self.get_object_value(var_oid)
                response_vars.append((var_oid, value))
            except SNMPError as e:
                return self._create_error_response(pdu, str(e), 1)
        
        return self._create_response(pdu, response_vars)
    
    def process_getnext_request(self, pdu):
        """处理GETNEXT请求"""
        response_vars = []
        
        for var_oid, _ in pdu['variables']:
            try:
                next_oid = self.find_next_object(var_oid)
                if next_oid:
                    value = self.get_object_value(next_oid)
                    response_vars.append((next_oid, value))
                else:
                    return self._create_error_response(pdu, "noSuchName", 1)
            except SNMPError as e:
                return self._create_error_response(pdu, str(e), 1)
        
        return self._create_response(pdu, response_vars)
    
    def process_set_request(self, pdu):
        """处理SET请求"""
        # 首先验证所有变量
        for i, (var_oid, value) in enumerate(pdu['variables']):
            if var_oid not in self.mib:
                return self._create_error_response(pdu, "noSuchName", i + 1)
        
        # 执行设置操作
        response_vars = []
        for i, (var_oid, value) in enumerate(pdu['variables']):
            try:
                self.set_object_value(var_oid, value)
                response_vars.append((var_oid, value))
            except SNMPError as e:
                return self._create_error_response(pdu, str(e), i + 1)
        
        return self._create_response(pdu, response_vars)
    
    def run(self):
        """运行代理"""
        self.running = True
        print(f"SNMP Agent listening on {self.host}:{self.port}")
        
        while self.running:
            try:
                data, addr = self.socket.recvfrom(65535)
                
                # 解码请求
                request = self._decode_message(data)
                
                # 验证团体名
                if request['community'] != self.community:
                    continue
                
                # 处理请求
                pdu = request['pdu']
                if pdu['type'] == 'get':
                    response = self.process_get_request(pdu)
                elif pdu['type'] == 'getnext':
                    response = self.process_getnext_request(pdu)
                elif pdu['type'] == 'set':
                    response = self.process_set_request(pdu)
                else:
                    continue
                
                # 发送响应
                response_data = self._encode_message(response)
                self.socket.sendto(response_data, addr)
                
            except Exception as e:
                print(f"Error processing request: {e}")
    
    def stop(self):
        """停止代理"""
        self.running = False
        self.socket.close()

# 使用示例
def run_agent():
    agent = SNMPAgent()
    
    # 注册自定义对象
    agent.register_object("1.3.6.1.4.1.12345.1.1.0", "Custom Object", "read-write")
    agent.register_object("1.3.6.1.4.1.12345.1.2.0", lambda: time.time(), "read-only")
    
    try:
        agent.run()
    except KeyboardInterrupt:
        print("\nStopping agent...")
        agent.stop()

if __name__ == "__main__":
    run_agent()

OwnerString ::= DisplayString
EntryStatus ::= INTEGER {
    valid(1),
    createRequest(2),
    underCreation(3),
    invalid(4)
}

createRequest(2) → underCreation(3) → valid(1)
                                   ↓
                              invalid(4)

etherStatsTable OBJECT-TYPE
    SYNTAX      SEQUENCE OF EtherStatsEntry
    ACCESS      not-accessible
    STATUS      mandatory
    DESCRIPTION "A list of Ethernet statistics entries."
    ::= { statistics 1 }

etherStatsEntry OBJECT-TYPE
    SYNTAX      EtherStatsEntry
    ACCESS      not-accessible
    STATUS      mandatory
    INDEX       { etherStatsIndex }
    ::= { etherStatsTable 1 }

EtherStatsEntry ::= SEQUENCE {
    etherStatsIndex             INTEGER,
    etherStatsDataSource        OBJECT IDENTIFIER,
    etherStatsDropEvents        Counter,
    etherStatsOctets           Counter,
    etherStatsPkts             Counter,
    etherStatsBroadcastPkts    Counter,
    etherStatsMulticastPkts    Counter,
    etherStatsCRCAlignErrors   Counter,
    etherStatsUndersizePkts    Counter,
    etherStatsOversizePkts     Counter,
    etherStatsFragments        Counter,
    etherStatsJabbers          Counter,
    etherStatsCollisions       Counter,
    etherStatsPkts64Octets     Counter,
    etherStatsPkts65to127Octets Counter,
    etherStatsPkts128to255Octets Counter,
    etherStatsPkts256to511Octets Counter,
    etherStatsPkts512to1023Octets Counter,
    etherStatsPkts1024to1518Octets Counter,
    etherStatsOwner            OwnerString,
    etherStatsStatus           EntryStatus
}

historyControlTable OBJECT-TYPE
    SYNTAX      SEQUENCE OF HistoryControlEntry
    ACCESS      not-accessible
    STATUS      mandatory
    ::= { history 1 }

HistoryControlEntry ::= SEQUENCE {
    historyControlIndex         INTEGER,
    historyControlDataSource    OBJECT IDENTIFIER,
    historyControlBucketsRequested INTEGER,
    historyControlBucketsGranted INTEGER,
    historyControlInterval      INTEGER,
    historyControlOwner         OwnerString,
    historyControlStatus        EntryStatus
}

etherHistoryTable OBJECT-TYPE
    SYNTAX      SEQUENCE OF EtherHistoryEntry
    ACCESS      not-accessible
    STATUS      mandatory
    ::= { history 2 }

EtherHistoryEntry ::= SEQUENCE {
    etherHistoryIndex           INTEGER,
    etherHistorySampleIndex     INTEGER,
    etherHistoryIntervalStart   TimeTicks,
    etherHistoryDropEvents      Counter,
    etherHistoryOctets         Counter,
    etherHistoryPkts           Counter,
    etherHistoryBroadcastPkts  Counter,
    etherHistoryMulticastPkts  Counter,
    etherHistoryCRCAlignErrors Counter,
    etherHistoryUndersizePkts  Counter,
    etherHistoryOversizePkts   Counter,
    etherHistoryFragments      Counter,
    etherHistoryJabbers        Counter,
    etherHistoryCollisions     Counter,
    etherHistoryUtilization    INTEGER
}

alarmTable OBJECT-TYPE
    SYNTAX      SEQUENCE OF AlarmEntry
    ACCESS      not-accessible
    STATUS      mandatory
    ::= { alarm 1 }

AlarmEntry ::= SEQUENCE {
    alarmIndex                  INTEGER,
    alarmInterval               INTEGER,
    alarmVariable               OBJECT IDENTIFIER,
    alarmSampleType             INTEGER,
    alarmValue                  INTEGER,
    alarmStartupAlarm           INTEGER,
    alarmRisingThreshold        INTEGER,
    alarmFallingThreshold       INTEGER,
    alarmRisingEventIndex       INTEGER,
    alarmFallingEventIndex      INTEGER,
    alarmOwner                  OwnerString,
    alarmStatus                 EntryStatus
}

hostTable OBJECT-TYPE
    SYNTAX      SEQUENCE OF HostEntry
    ACCESS      not-accessible
    STATUS      mandatory
    ::= { hosts 4 }

HostEntry ::= SEQUENCE {
    hostAddress                 OCTET STRING,
    hostCreationOrder          INTEGER,
    hostIndex                  INTEGER,
    hostInPkts                 Counter,
    hostOutPkts                Counter,
    hostInOctets               Counter,
    hostOutOctets              Counter,
    hostOutErrors              Counter,
    hostOutBroadcastPkts       Counter,
    hostOutMulticastPkts       Counter
}

matrixSDTable OBJECT-TYPE
    SYNTAX      SEQUENCE OF MatrixSDEntry
    ACCESS      not-accessible
    STATUS      mandatory
    ::= { matrix 1 }

MatrixSDEntry ::= SEQUENCE {
    matrixSDSourceAddress       OCTET STRING,
    matrixSDDestAddress         OCTET STRING,
    matrixSDIndex              INTEGER,
    matrixSDPkts               Counter,
    matrixSDOctets             Counter,
    matrixSDErrors             Counter
}

import time
import threading
from collections import defaultdict, deque
from dataclasses import dataclass
from typing import Dict, List, Optional

@dataclass
class EthernetStats:
    """以太网统计信息"""
    octets: int = 0
    pkts: int = 0
    broadcast_pkts: int = 0
    multicast_pkts: int = 0
    crc_align_errors: int = 0
    undersize_pkts: int = 0
    oversize_pkts: int = 0
    fragments: int = 0
    jabbers: int = 0
    collisions: int = 0
    pkts_64_octets: int = 0
    pkts_65_to_127_octets: int = 0
    pkts_128_to_255_octets: int = 0
    pkts_256_to_511_octets: int = 0
    pkts_512_to_1023_octets: int = 0
    pkts_1024_to_1518_octets: int = 0
    drop_events: int = 0

class RMONStatisticsCollector:
    """RMON统计信息收集器"""
    
    def __init__(self):
        self.interfaces = {}  # interface_index -> EthernetStats
        self.running = False
        self.collection_thread = None
        self.lock = threading.Lock()
    
    def add_interface(self, interface_index: int, data_source: str):
        """添加监控接口"""
        with self.lock:
            self.interfaces[interface_index] = {
                'data_source': data_source,
                'stats': EthernetStats(),
                'owner': 'monitor',
                'status': 'valid'
            }
    
    def remove_interface(self, interface_index: int):
        """移除监控接口"""
        with self.lock:
            if interface_index in self.interfaces:
                del self.interfaces[interface_index]
    
    def update_packet_stats(self, interface_index: int, packet_size: int, 
                          packet_type: str, has_error: bool = False):
        """更新分组统计信息"""
        with self.lock:
            if interface_index not in self.interfaces:
                return
            
            stats = self.interfaces[interface_index]['stats']
            
            # 更新基本统计
            stats.octets += packet_size
            stats.pkts += 1
            
            # 根据分组类型更新
            if packet_type == 'broadcast':
                stats.broadcast_pkts += 1
            elif packet_type == 'multicast':
                stats.multicast_pkts += 1
            
            # 根据分组大小分类
            if packet_size == 64:
                stats.pkts_64_octets += 1
            elif 65 <= packet_size <= 127:
                stats.pkts_65_to_127_octets += 1
            elif 128 <= packet_size <= 255:
                stats.pkts_128_to_255_octets += 1
            elif 256 <= packet_size <= 511:
                stats.pkts_256_to_511_octets += 1
            elif 512 <= packet_size <= 1023:
                stats.pkts_512_to_1023_octets += 1
            elif 1024 <= packet_size <= 1518:
                stats.pkts_1024_to_1518_octets += 1
            elif packet_size > 1518:
                stats.oversize_pkts += 1
            elif packet_size < 64:
                stats.undersize_pkts += 1
            
            # 错误统计
            if has_error:
                stats.crc_align_errors += 1
    
    def get_interface_stats(self, interface_index: int) -> Optional[EthernetStats]:
        """获取接口统计信息"""
        with self.lock:
            if interface_index in self.interfaces:
                return self.interfaces[interface_index]['stats']
        return None
    
    def get_all_stats(self) -> Dict[int, EthernetStats]:
        """获取所有接口统计信息"""
        with self.lock:
            return {idx: info['stats'] for idx, info in self.interfaces.items()}

# 使用示例
def demo_rmon_statistics():
    """RMON统计演示"""
    collector = RMONStatisticsCollector()
    
    # 添加监控接口
    collector.add_interface(1, "eth0")
    collector.add_interface(2, "eth1")
    
    # 模拟分组统计
    import random
    for _ in range(1000):
        interface = random.choice([1, 2])
        size = random.randint(64, 1518)
        pkt_type = random.choice(['unicast', 'broadcast', 'multicast'])
        has_error = random.random() < 0.01  # 1%错误率
        
        collector.update_packet_stats(interface, size, pkt_type, has_error)
    
    # 获取统计结果
    for interface_idx in [1, 2]:
        stats = collector.get_interface_stats(interface_idx)
        if stats:
            print(f"接口 {interface_idx} 统计:")
            print(f"  总字节数: {stats.octets}")
            print(f"  总分组数: {stats.pkts}")
            print(f"  广播分组: {stats.broadcast_pkts}")
            print(f"  组播分组: {stats.multicast_pkts}")
            print(f"  CRC错误: {stats.crc_align_errors}")
            print()

if __name__ == "__main__":
    demo_rmon_statistics()

import time
from collections import deque
from dataclasses import dataclass, asdict
from typing import Dict, List, Optional

@dataclass
class HistorySample:
    """历史采样数据"""
    sample_index: int
    interval_start: int
    drop_events: int
    octets: int
    pkts: int
    broadcast_pkts: int
    multicast_pkts: int
    crc_align_errors: int
    undersize_pkts: int
    oversize_pkts: int
    fragments: int
    jabbers: int
    collisions: int
    utilization: int  # 百分比

class RMONHistoryCollector:
    """RMON历史数据收集器"""
    
    def __init__(self):
        self.history_controls = {}  # control_index -> control_info
        self.history_data = {}      # control_index -> deque of samples
        self.running = False
        self.collection_thread = None
    
    def create_history_control(self, control_index: int, data_source: str,
                             buckets_requested: int = 50, interval: int = 1800):
        """创建历史控制条目"""
        self.history_controls[control_index] = {
            'data_source': data_source,
            'buckets_requested': buckets_requested,
            'buckets_granted': min(buckets_requested, 65535),
            'interval': interval,  # 秒
            'owner': 'monitor',
            'status': 'valid',
            'last_sample_time': 0,
            'sample_counter': 0
        }
        
        # 初始化历史数据队列
        max_samples = self.history_controls[control_index]['buckets_granted']
        self.history_data[control_index] = deque(maxlen=max_samples)
    
    def collect_sample(self, control_index: int, stats: EthernetStats):
        """收集历史采样"""
        if control_index not in self.history_controls:
            return
        
        control = self.history_controls[control_index]
        current_time = int(time.time())
        
        # 检查是否到了采样时间
        if current_time - control['last_sample_time'] < control['interval']:
            return
        
        # 创建新的采样
        control['sample_counter'] += 1
        sample = HistorySample(
            sample_index=control['sample_counter'],
            interval_start=current_time,
            drop_events=stats.drop_events,
            octets=stats.octets,
            pkts=stats.pkts,
            broadcast_pkts=stats.broadcast_pkts,
            multicast_pkts=stats.multicast_pkts,
            crc_align_errors=stats.crc_align_errors,
            undersize_pkts=stats.undersize_pkts,
            oversize_pkts=stats.oversize_pkts,
            fragments=stats.fragments,
            jabbers=stats.jabbers,
            collisions=stats.collisions,
            utilization=self._calculate_utilization(stats, control['interval'])
        )
        
        # 添加到历史数据队列
        self.history_data[control_index].append(sample)
        control['last_sample_time'] = current_time
    
    def _calculate_utilization(self, stats: EthernetStats, interval: int) -> int:
        """计算链路利用率"""
        # 假设10Mbps以太网
        max_bits_per_second = 10 * 1000 * 1000
        actual_bits = stats.octets * 8
        utilization = (actual_bits / (max_bits_per_second * interval)) * 100
        return min(int(utilization), 100)
    
    def get_history_samples(self, control_index: int, 
                          start_sample: int = None, 
                          end_sample: int = None) -> List[HistorySample]:
        """获取历史采样数据"""
        if control_index not in self.history_data:
            return []
        
        samples = list(self.history_data[control_index])
        
        if start_sample is not None:
            samples = [s for s in samples if s.sample_index >= start_sample]
        
        if end_sample is not None:
            samples = [s for s in samples if s.sample_index <= end_sample]
        
        return samples
    
    def get_latest_samples(self, control_index: int, count: int = 10) -> List[HistorySample]:
        """获取最新的N个采样"""
        if control_index not in self.history_data:
            return []
        
        samples = list(self.history_data[control_index])
        return samples[-count:] if len(samples) >= count else samples

# 使用示例
def demo_rmon_history():
    """RMON历史数据演示"""
    history_collector = RMONHistoryCollector()
    stats_collector = RMONStatisticsCollector()
    
    # 创建历史控制
    history_collector.create_history_control(
        control_index=1,
        data_source="eth0",
        buckets_requested=24,  # 24小时
        interval=3600  # 每小时采样
    )
    
    # 添加统计接口
    stats_collector.add_interface(1, "eth0")
    
    # 模拟数据收集
    for hour in range(24):
        # 模拟一小时的流量
        for _ in range(1000):
            size = random.randint(64, 1518)
            pkt_type = random.choice(['unicast', 'broadcast', 'multicast'])
            stats_collector.update_packet_stats(1, size, pkt_type)
        
        # 收集历史采样
        current_stats = stats_collector.get_interface_stats(1)
        if current_stats:
            history_collector.collect_sample(1, current_stats)
        
        # 模拟时间推进
        time.sleep(0.1)  # 实际应用中是3600秒
    
    # 获取历史数据
    samples = history_collector.get_latest_samples(1, 10)
    print("最近10个历史采样:")
    for sample in samples:
        print(f"采样 {sample.sample_index}: "
              f"分组数={sample.pkts}, "
              f"字节数={sample.octets}, "
              f"利用率={sample.utilization}%")

if __name__ == "__main__":
    demo_rmon_history()

import time
import threading
from enum import Enum
from dataclasses import dataclass
from typing import Dict, List, Callable, Optional

class AlarmSampleType(Enum):
    ABSOLUTE_VALUE = 1
    DELTA_VALUE = 2

class AlarmStartup(Enum):
    RISING = 1
    FALLING = 2
    RISING_OR_FALLING = 3

@dataclass
class AlarmEntry:
    """报警条目"""
    alarm_index: int
    interval: int
    variable: str  # OID
    sample_type: AlarmSampleType
    rising_threshold: int
    falling_threshold: int
    rising_event_index: int
    falling_event_index: int
    startup_alarm: AlarmStartup
    owner: str
    status: str
    last_sample_value: Optional[int] = None
    last_sample_time: int = 0
    alarm_state: str = "normal"  # normal, rising, falling

class RMONAlarmManager:
    """RMON报警管理器"""
    
    def __init__(self, event_manager):
        self.alarms = {}  # alarm_index -> AlarmEntry
        self.event_manager = event_manager
        self.running = False
        self.alarm_thread = None
        self.lock = threading.Lock()
    
    def create_alarm(self, alarm_index: int, interval: int, variable: str,
                    sample_type: AlarmSampleType, rising_threshold: int,
                    falling_threshold: int, rising_event_index: int,
                    falling_event_index: int, startup_alarm: AlarmStartup,
                    owner: str = "monitor"):
        """创建报警条目"""
        with self.lock:
            alarm = AlarmEntry(
                alarm_index=alarm_index,
                interval=interval,
                variable=variable,
                sample_type=sample_type,
                rising_threshold=rising_threshold,
                falling_threshold=falling_threshold,
                rising_event_index=rising_event_index,
                falling_event_index=falling_event_index,
                startup_alarm=startup_alarm,
                owner=owner,
                status="valid"
            )
            self.alarms[alarm_index] = alarm
    
    def delete_alarm(self, alarm_index: int):
        """删除报警条目"""
        with self.lock:
            if alarm_index in self.alarms:
                del self.alarms[alarm_index]
    
    def check_alarm(self, alarm_index: int, current_value: int):
        """检查报警条件"""
        with self.lock:
            if alarm_index not in self.alarms:
                return
            
            alarm = self.alarms[alarm_index]
            current_time = int(time.time())
            
            # 检查采样间隔
            if current_time - alarm.last_sample_time < alarm.interval:
                return
            
            # 计算采样值
            if alarm.sample_type == AlarmSampleType.ABSOLUTE_VALUE:
                sample_value = current_value
            else:  # DELTA_VALUE
                if alarm.last_sample_value is not None:
                    sample_value = current_value - alarm.last_sample_value
                else:
                    sample_value = 0
            
            # 检查阈值
            self._check_thresholds(alarm, sample_value)
            
            # 更新采样信息
            alarm.last_sample_value = current_value
            alarm.last_sample_time = current_time
    
    def _check_thresholds(self, alarm: AlarmEntry, sample_value: int):
        """检查阈值条件"""
        # 上升阈值检查
        if (sample_value >= alarm.rising_threshold and 
            alarm.alarm_state != "rising"):
            
            alarm.alarm_state = "rising"
            self.event_manager.trigger_event(
                alarm.rising_event_index,
                f"Rising alarm on {alarm.variable}: {sample_value} >= {alarm.rising_threshold}"
            )
        
        # 下降阈值检查
        elif (sample_value <= alarm.falling_threshold and 
              alarm.alarm_state != "falling"):
            
            alarm.alarm_state = "falling"
            self.event_manager.trigger_event(
                alarm.falling_event_index,
                f"Falling alarm on {alarm.variable}: {sample_value} <= {alarm.falling_threshold}"
            )
        
        # 正常状态
        elif (alarm.falling_threshold < sample_value < alarm.rising_threshold):
            alarm.alarm_state = "normal"
    
    def start_monitoring(self):
        """开始监控"""
        self.running = True
        self.alarm_thread = threading.Thread(target=self._monitor_loop)
        self.alarm_thread.start()
    
    def stop_monitoring(self):
        """停止监控"""
        self.running = False
        if self.alarm_thread:
            self.alarm_thread.join()
    
    def _monitor_loop(self):
        """监控循环"""
        while self.running:
            # 这里应该从实际的MIB对象获取值
            # 为演示目的，我们使用模拟值
            for alarm_index in list(self.alarms.keys()):
                # 模拟获取变量值
                current_value = self._get_variable_value(
                    self.alarms[alarm_index].variable
                )
                if current_value is not None:
                    self.check_alarm(alarm_index, current_value)
            
            time.sleep(1)
    
    def _get_variable_value(self, variable: str) -> Optional[int]:
        """获取变量值（模拟实现）"""
        # 实际实现中应该从SNMP代理获取值
        import random
        return random.randint(0, 1000)

class RMONEventManager:
    """RMON事件管理器"""
    
    def __init__(self):
        self.events = {}  # event_index -> event_info
        self.event_log = []
        self.lock = threading.Lock()
    
    def create_event(self, event_index: int, event_type: str,
                    description: str, owner: str = "monitor"):
        """创建事件条目"""
        with self.lock:
            self.events[event_index] = {
                'type': event_type,  # log, snmp-trap, log-and-trap
                'description': description,
                'owner': owner,
                'status': 'valid'
            }
    
    def trigger_event(self, event_index: int, message: str):
        """触发事件"""
        with self.lock:
            if event_index not in self.events:
                return
            
            event = self.events[event_index]
            timestamp = int(time.time())
            
            # 记录事件日志
            log_entry = {
                'event_index': event_index,
                'timestamp': timestamp,
                'message': message,
                'description': event['description']
            }
            self.event_log.append(log_entry)
            
            # 根据事件类型执行相应动作
            if event['type'] in ['log', 'log-and-trap']:
                self._log_event(log_entry)
            
            if event['type'] in ['snmp-trap', 'log-and-trap']:
                self._send_trap(log_entry)
    
    def _log_event(self, log_entry: dict):
        """记录事件到日志"""
        print(f"[{log_entry['timestamp']}] Event {log_entry['event_index']}: "
              f"{log_entry['message']}")
    
    def _send_trap(self, log_entry: dict):
        """发送SNMP陷阱"""
        # 实际实现中应该发送SNMP Trap
        print(f"TRAP: Event {log_entry['event_index']} - {log_entry['message']}")
    
    def get_event_log(self, max_entries: int = 100) -> List[dict]:
        """获取事件日志"""
        with self.lock:
            return self.event_log[-max_entries:]

# 使用示例
def demo_rmon_alarm():
    """RMON报警演示"""
    event_manager = RMONEventManager()
    alarm_manager = RMONAlarmManager(event_manager)
    
    # 创建事件
    event_manager.create_event(1, "log-and-trap", "High traffic alarm")
    event_manager.create_event(2, "log", "Low traffic alarm")
    
    # 创建报警
    alarm_manager.create_alarm(
        alarm_index=1,
        interval=30,  # 30秒检查一次
        variable="1.3.6.1.2.1.2.2.1.10.1",  # ifInOctets.1
        sample_type=AlarmSampleType.DELTA_VALUE,
        rising_threshold=1000000,  # 1MB/30s
        falling_threshold=100000,   # 100KB/30s
        rising_event_index=1,
        falling_event_index=2,
        startup_alarm=AlarmStartup.RISING_OR_FALLING
    )
    
    # 开始监控
    alarm_manager.start_monitoring()
    
    # 运行一段时间
    time.sleep(10)
    
    # 停止监控
    alarm_manager.stop_monitoring()
    
    # 查看事件日志
    log_entries = event_manager.get_event_log()
    print(f"\n事件日志 ({len(log_entries)} 条):")
    for entry in log_entries:
        print(f"  {entry['timestamp']}: {entry['message']}")

if __name__ == "__main__":
    demo_rmon_alarm()

ipconfig [/all] [/renew[Adapter]] [/release[Adapter]] [/flushdns] 
         [/displaydns] [/registerdns] [/showclassid Adapter] 
         [/setclassid Adapter [ClassID]]

# 显示基本网络配置
ipconfig

# 显示详细网络配置
ipconfig /all

# 释放并重新获取IP地址
ipconfig /release
ipconfig /renew

# 清空DNS缓存
ipconfig /flushdns

ping [-t] [-a] [-n Count] [-l Size] [-f] [-i TTL] [-v TOS] 
     [-r Count] [-s Count] [{-j HostList | -k HostList}] 
     [-w Timeout] [TargetName]

# 基本ping测试
ping www.baidu.com

# 持续ping
ping -t 192.168.1.1

# 发送10个大包
ping -n 10 -l 1024 192.168.1.1

# 设置超时时间
ping -w 3000 192.168.1.1

arp [-a[InetAddr] [-N IfaceAddr]] [-g [InetAddr][-N IfaceAddr]] 
    [-d InetAddr[IfaceAddr]] [-s InetAddr EtherAddr[IfaceAddr]]

# 显示所有ARP表项
arp -a

# 显示指定IP的ARP表项
arp -a 192.168.1.1

# 删除ARP表项
arp -d 192.168.1.1

# 添加静态ARP表项
arp -s 192.168.1.100 00-11-22-33-44-55

netstat [-a] [-e] [-n] [-o] [-p Protocol] [-r] [-s] [Interval]

# 显示所有网络连接
netstat -a

# 显示TCP连接及进程ID
netstat -ano -p tcp

# 显示路由表
netstat -r

# 显示网络统计信息
netstat -s

tracert [-d] [-h MaximumHops] [-j HostList] [-w Timeout] [TargetName]

# 基本路由跟踪
tracert www.baidu.com

# 不解析主机名
tracert -d 8.8.8.8

# 设置最大跳数
tracert -h 15 www.google.com

pathping [-n] [-h MaxHops] [-w Timeout] [Target]

pathping www.google.com    # 分析路径延迟和丢包
pathping -n 192.168.1.1    # 禁用主机名解析

route print                  # 显示当前路由表
route add 目标网络 mask 子网掩码 网关  # 添加静态路由
route delete 目标网络        # 删除路由

route print                              # 查看路由表
route add 10.0.0.0 mask 255.0.0.0 192.168.1.1  # 添加静态路由
route delete 10.0.0.0                    # 删除路由

nslookup [域名/IP] [DNS服务器]

nslookup www.baidu.com          # 查询A记录
nslookup -type=mx example.com   # 查询MX记录（邮件服务器）
nslookup 8.8.8.8               # 反向DNS查询

nslookup
> server 8.8.8.8     # 切换DNS服务器
> set type=AAAA      # 查询IPv6记录
> www.google.com

import socket
import struct
import threading
import time
from dataclasses import dataclass
from typing import Dict, List, Optional
from collections import defaultdict, deque

@dataclass
class PacketInfo:
    """数据包信息"""
    timestamp: float
    src_mac: str
    dst_mac: str
    src_ip: str
    dst_ip: str
    protocol: str
    size: int
    src_port: Optional[int] = None
    dst_port: Optional[int] = None

class NetworkSniffer:
    """网络嗅探器"""
    
    def __init__(self, interface: str = None):
        self.interface = interface
        self.running = False
        self.packets = deque(maxlen=10000)  # 保存最近10000个包
        self.statistics = defaultdict(int)
        self.protocol_stats = defaultdict(int)
        self.size_stats = defaultdict(int)
        self.lock = threading.Lock()
        
        # 创建原始套接字
        try:
            self.socket = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.ntohs(0x0003))
            if interface:
                self.socket.bind((interface, 0))
        except PermissionError:
            raise PermissionError("需要管理员权限来创建原始套接字")
        except Exception as e:
            raise Exception(f"创建套接字失败: {e}")
    
    def start_capture(self):
        """开始捕获"""
        self.running = True
        self.capture_thread = threading.Thread(target=self._capture_loop)
        self.capture_thread.start()
    
    def stop_capture(self):
        """停止捕获"""
        self.running = False
        if hasattr(self, 'capture_thread'):
            self.capture_thread.join()
        self.socket.close()
    
    def _capture_loop(self):
        """捕获循环"""
        while self.running:
            try:
                # 接收数据包
                packet, addr = self.socket.recvfrom(65535)
                timestamp = time.time()
                
                # 解析数据包
                packet_info = self._parse_packet(packet, timestamp)
                if packet_info:
                    with self.lock:
                        self.packets.append(packet_info)
                        self._update_statistics(packet_info)
                
            except Exception as e:
                if self.running:
                    print(f"捕获数据包时出错: {e}")
                break
    
    def _parse_packet(self, packet: bytes, timestamp: float) -> Optional[PacketInfo]:
        """解析数据包"""
        try:
            # 解析以太网头部
            eth_header = packet[:14]
            eth = struct.unpack('!6s6sH', eth_header)
            
            dst_mac = ':'.join(['%02x' % b for b in eth[0]])
            src_mac = ':'.join(['%02x' % b for b in eth[1]])
            eth_type = eth[2]
            
            # 只处理IP数据包
            if eth_type != 0x0800:
                return None
            
            # 解析IP头部
            ip_header = packet[14:34]
            ip = struct.unpack('!BBHHHBBH4s4s', ip_header)
            
            version = ip[0] >> 4
            ihl = ip[0] & 0xF
            protocol = ip[6]
            src_ip = socket.inet_ntoa(ip[8])
            dst_ip = socket.inet_ntoa(ip[9])
            
            # 解析传输层协议
            src_port = dst_port = None
            protocol_name = 'Unknown'
            
            if protocol == 6:  # TCP
                protocol_name = 'TCP'
                tcp_header = packet[34:54]
                tcp = struct.unpack('!HHLLBBHHH', tcp_header)
                src_port = tcp[0]
                dst_port = tcp[1]
            elif protocol == 17:  # UDP
                protocol_name = 'UDP'
                udp_header = packet[34:42]
                udp = struct.unpack('!HHHH', udp_header)
                src_port = udp[0]
                dst_port = udp[1]
            elif protocol == 1:  # ICMP
                protocol_name = 'ICMP'
            
            return PacketInfo(
                timestamp=timestamp,
                src_mac=src_mac,
                dst_mac=dst_mac,
                src_ip=src_ip,
                dst_ip=dst_ip,
                protocol=protocol_name,
                size=len(packet),
                src_port=src_port,
                dst_port=dst_port
            )
            
        except Exception as e:
            return None
    
    def _update_statistics(self, packet: PacketInfo):
        """更新统计信息"""
        self.statistics['total_packets'] += 1
        self.statistics['total_bytes'] += packet.size
        self.protocol_stats[packet.protocol] += 1
        
        # 按大小分类
        if packet.size <= 64:
            self.size_stats['64'] += 1
        elif packet.size <= 128:
            self.size_stats['65-128'] += 1
        elif packet.size <= 256:
            self.size_stats['129-256'] += 1
        elif packet.size <= 512:
            self.size_stats['257-512'] += 1
        elif packet.size <= 1024:
            self.size_stats['513-1024'] += 1
        else:
            self.size_stats['1024+'] += 1
        
        # 检查广播和组播
        if packet.dst_mac.startswith('ff:ff:ff'):
            self.statistics['broadcast_packets'] += 1
        elif packet.dst_mac.startswith('01:00:5e'):
            self.statistics['multicast_packets'] += 1
    
    def get_statistics(self) -> dict:
        """获取统计信息"""
        with self.lock:
            return {
                'general': dict(self.statistics),
                'protocols': dict(self.protocol_stats),
                'sizes': dict(self.size_stats),
                'packet_count': len(self.packets)
            }
    
    def get_recent_packets(self, count: int = 100) -> List[PacketInfo]:
        """获取最近的数据包"""
        with self.lock:
            return list(self.packets)[-count:]
    
    def filter_packets(self, **filters) -> List[PacketInfo]:
        """过滤数据包"""
        with self.lock:
            filtered = []
            for packet in self.packets:
                match = True
                
                if 'src_ip' in filters and packet.src_ip != filters['src_ip']:
                    match = False
                if 'dst_ip' in filters and packet.dst_ip != filters['dst_ip']:
                    match = False
                if 'protocol' in filters and packet.protocol != filters['protocol']:
                    match = False
                if 'src_port' in filters and packet.src_port != filters['src_port']:
                    match = False
                if 'dst_port' in filters and packet.dst_port != filters['dst_port']:
                    match = False
                
                if match:
                    filtered.append(packet)
            
            return filtered
    
    def get_top_talkers(self, count: int = 10) -> List[tuple]:
        """获取流量最大的主机"""
        ip_stats = defaultdict(int)
        
        with self.lock:
            for packet in self.packets:
                ip_stats[packet.src_ip] += packet.size
                ip_stats[packet.dst_ip] += packet.size
        
        return sorted(ip_stats.items(), key=lambda x: x[1], reverse=True)[:count]
    
    def get_protocol_distribution(self) -> dict:
        """获取协议分布"""
        with self.lock:
            total = sum(self.protocol_stats.values())
            if total == 0:
                return {}
            
            return {proto: (count / total) * 100 
                   for proto, count in self.protocol_stats.items()}

class NetworkAnalyzer:
    """网络分析器"""
    
    def __init__(self, sniffer: NetworkSniffer):
        self.sniffer = sniffer
    
    def analyze_traffic_patterns(self) -> dict:
        """分析流量模式"""
        packets = self.sniffer.get_recent_packets(1000)
        
        if not packets:
            return {}
        
        # 时间分布分析
        time_buckets = defaultdict(int)
        for packet in packets:
            bucket = int(packet.timestamp) // 60  # 按分钟分组
            time_buckets[bucket] += 1
        
        # 流量方向分析
        internal_ips = set()
        external_ips = set()
        
        for packet in packets:
            if packet.src_ip.startswith('192.168.') or packet.src_ip.startswith('10.'):
                internal_ips.add(packet.src_ip)
            else:
                external_ips.add(packet.src_ip)
            
            if packet.dst_ip.startswith('192.168.') or packet.dst_ip.startswith('10.'):
                internal_ips.add(packet.dst_ip)
            else:
                external_ips.add(packet.dst_ip)
        
        return {
            'time_distribution': dict(time_buckets),
            'internal_hosts': len(internal_ips),
            'external_hosts': len(external_ips),
            'total_packets': len(packets)
        }
    
    def detect_anomalies(self) -> List[dict]:
        """检测异常"""
        anomalies = []
        stats = self.sniffer.get_statistics()
        
        # 检测异常高的流量
        if stats['general'].get('total_packets', 0) > 10000:
            anomalies.append({
                'type': 'high_traffic',
                'description': '检测到异常高的数据包数量',
                'value': stats['general']['total_packets']
            })
        
        # 检测异常的协议分布
        protocol_dist = self.sniffer.get_protocol_distribution()
        if protocol_dist.get('ICMP', 0) > 50:
            anomalies.append({
                'type': 'icmp_flood',
                'description': 'ICMP流量异常高',
                'value': f"{protocol_dist['ICMP']:.1f}%"
            })
        
        return anomalies

# 使用示例（需要管理员权限）
def demo_network_sniffer():
    """网络嗅探器演示"""
    try:
        # 创建嗅探器
        sniffer = NetworkSniffer()
        analyzer = NetworkAnalyzer(sniffer)
        
        print("开始网络监听...")
        sniffer.start_capture()
        
        # 运行一段时间
        time.sleep(30)
        
        # 获取统计信息
        stats = sniffer.get_statistics()
        print(f"\n统计信息:")
        print(f"  总数据包: {stats['general'].get('total_packets', 0)}")
        print(f"  总字节数: {stats['general'].get('total_bytes', 0)}")
        print(f"  广播包: {stats['general'].get('broadcast_packets', 0)}")
        
        print(f"\n协议分布:")
        for proto, count in stats['protocols'].items():
            print(f"  {proto}: {count}")
        
        # 获取流量最大的主机
        top_talkers = sniffer.get_top_talkers(5)
        print(f"\n流量最大的主机:")
        for ip, bytes_count in top_talkers:
            print(f"  {ip}: {bytes_count} 字节")
        
        # 分析流量模式
        patterns = analyzer.analyze_traffic_patterns()
        print(f"\n流量分析:")
        print(f"  内部主机: {patterns.get('internal_hosts', 0)}")
        print(f"  外部主机: {patterns.get('external_hosts', 0)}")
        
        # 检测异常
        anomalies = analyzer.detect_anomalies()
        if anomalies:
            print(f"\n检测到异常:")
            for anomaly in anomalies:
                print(f"  {anomaly['description']}: {anomaly['value']}")
        
        # 停止监听
        sniffer.stop_capture()
        print("\n监听已停止")
        
    except PermissionError:
        print("错误: 需要管理员权限来运行网络嗅探器")
    except Exception as e:
        print(f"错误: {e}")

if __name__ == "__main__":
    demo_network_sniffer()

import asyncio
import json
import sqlite3
import time
from dataclasses import dataclass, asdict
from typing import Dict, List, Optional, Callable
from enum import Enum

class DeviceStatus(Enum):
    UP = "up"
    DOWN = "down"
    WARNING = "warning"
    UNKNOWN = "unknown"

@dataclass
class NetworkDevice:
    """网络设备"""
    id: str
    name: str
    ip_address: str
    device_type: str
    location: str
    status: DeviceStatus
    last_seen: float
    snmp_community: str = "public"
    snmp_port: int = 161

@dataclass
class MonitoringMetric:
    """监控指标"""
    device_id: str
    metric_name: str
    value: float
    unit: str
    timestamp: float
    threshold_warning: Optional[float] = None
    threshold_critical: Optional[float] = None

class NetworkManagementPlatform:
    """网络管理平台"""
    
    def __init__(self, db_path: str = "network_management.db"):
        self.db_path = db_path
        self.devices = {}
        self.metrics = []
        self.alerts = []
        self.monitoring_tasks = {}
        self.event_handlers = {}
        
        # 初始化数据库
        self._init_database()
    
    def _init_database(self):
        """初始化数据库"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # 创建设备表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS devices (
                id TEXT PRIMARY KEY,
                name TEXT NOT NULL,
                ip_address TEXT NOT NULL,
                device_type TEXT NOT NULL,
                location TEXT,
                status TEXT NOT NULL,
                last_seen REAL,
                snmp_community TEXT,
                snmp_port INTEGER,
                created_at REAL DEFAULT (julianday('now'))
            )
        ''')
        
        # 创建指标表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS metrics (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                device_id TEXT NOT NULL,
                metric_name TEXT NOT NULL,
                value REAL NOT NULL,
                unit TEXT,
                timestamp REAL NOT NULL,
                threshold_warning REAL,
                threshold_critical REAL,
                FOREIGN KEY (device_id) REFERENCES devices (id)
            )
        ''')
        
        # 创建告警表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS alerts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                device_id TEXT NOT NULL,
                alert_type TEXT NOT NULL,
                severity TEXT NOT NULL,
                message TEXT NOT NULL,
                timestamp REAL NOT NULL,
                acknowledged BOOLEAN DEFAULT FALSE,
                resolved BOOLEAN DEFAULT FALSE,
                FOREIGN KEY (device_id) REFERENCES devices (id)
            )
        ''')
        
        conn.commit()
        conn.close()
    
    def add_device(self, device: NetworkDevice):
        """添加设备"""
        self.devices[device.id] = device
        
        # 保存到数据库
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT OR REPLACE INTO devices 
            (id, name, ip_address, device_type, location, status, last_seen, snmp_community, snmp_port)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        ''', (
            device.id, device.name, device.ip_address, device.device_type,
            device.location, device.status.value, device.last_seen,
            device.snmp_community, device.snmp_port
        ))
        
        conn.commit()
        conn.close()
        
        # 启动监控
        self.start_device_monitoring(device.id)
    
    def remove_device(self, device_id: str):
        """移除设备"""
        if device_id in self.devices:
            del self.devices[device_id]
        
        # 停止监控
        self.stop_device_monitoring(device_id)
        
        # 从数据库删除
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('DELETE FROM devices WHERE id = ?', (device_id,))
        conn.commit()
        conn.close()
    
    def get_device(self, device_id: str) -> Optional[NetworkDevice]:
        """获取设备"""
        return self.devices.get(device_id)
    
    def get_all_devices(self) -> List[NetworkDevice]:
        """获取所有设备"""
        return list(self.devices.values())
    
    def update_device_status(self, device_id: str, status: DeviceStatus):
        """更新设备状态"""
        if device_id in self.devices:
            old_status = self.devices[device_id].status
            self.devices[device_id].status = status
            self.devices[device_id].last_seen = time.time()
            
            # 状态变化时生成告警
            if old_status != status:
                self._generate_status_alert(device_id, old_status, status)
    
    def add_metric(self, metric: MonitoringMetric):
        """添加监控指标"""
        self.metrics.append(metric)
        
        # 保存到数据库
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO metrics 
            (device_id, metric_name, value, unit, timestamp, threshold_warning, threshold_critical)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        ''', (
            metric.device_id, metric.metric_name, metric.value, metric.unit,
            metric.timestamp, metric.threshold_warning, metric.threshold_critical
        ))
        
        conn.commit()
        conn.close()
        
        # 检查阈值
        self._check_metric_thresholds(metric)
    
    def get_device_metrics(self, device_id: str, metric_name: str = None, 
                          hours: int = 24) -> List[MonitoringMetric]:
        """获取设备指标"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        since_time = time.time() - (hours * 3600)
        
        if metric_name:
            cursor.execute('''
                SELECT device_id, metric_name, value, unit, timestamp, threshold_warning, threshold_critical
                FROM metrics 
                WHERE device_id = ? AND metric_name = ? AND timestamp > ?
                ORDER BY timestamp DESC
            ''', (device_id, metric_name, since_time))
        else:
            cursor.execute('''
                SELECT device_id, metric_name, value, unit, timestamp, threshold_warning, threshold_critical
                FROM metrics 
                WHERE device_id = ? AND timestamp > ?
                ORDER BY timestamp DESC
            ''', (device_id, since_time))
        
        metrics = []
        for row in cursor.fetchall():
            metrics.append(MonitoringMetric(
                device_id=row[0],
                metric_name=row[1],
                value=row[2],
                unit=row[3],
                timestamp=row[4],
                threshold_warning=row[5],
                threshold_critical=row[6]
            ))
        
        conn.close()
        return metrics
    
    def start_device_monitoring(self, device_id: str):
        """开始设备监控"""
        if device_id not in self.monitoring_tasks:
            task = asyncio.create_task(self._monitor_device(device_id))
            self.monitoring_tasks[device_id] = task
    
    def stop_device_monitoring(self, device_id: str):
        """停止设备监控"""
        if device_id in self.monitoring_tasks:
            self.monitoring_tasks[device_id].cancel()
            del self.monitoring_tasks[device_id]
    
    async def _monitor_device(self, device_id: str):
        """监控设备"""
        while True:
            try:
                device = self.devices.get(device_id)
                if not device:
                    break
                
                # 执行ping测试
                ping_success = await self._ping_device(device.ip_address)
                
                if ping_success:
                    self.update_device_status(device_id, DeviceStatus.UP)
                    
                    # 收集SNMP指标
                    await self._collect_snmp_metrics(device)
                else:
                    self.update_device_status(device_id, DeviceStatus.DOWN)
                
                # 等待下次监控
                await asyncio.sleep(60)  # 每分钟监控一次
                
            except asyncio.CancelledError:
                break
            except Exception as e:
                print(f"监控设备 {device_id} 时出错: {e}")
                await asyncio.sleep(60)
    
    async def _ping_device(self, ip_address: str) -> bool:
        """Ping设备"""
        try:
            process = await asyncio.create_subprocess_exec(
                'ping', '-n', '1', '-w', '3000', ip_address,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE
            )
            
            stdout, stderr = await process.communicate()
            return process.returncode == 0
        except Exception:
            return False
    
    async def _collect_snmp_metrics(self, device: NetworkDevice):
        """收集SNMP指标"""
        try:
            # 这里应该实现实际的SNMP查询
            # 为演示目的，我们生成模拟数据
            import random
            
            timestamp = time.time()
            
            # CPU利用率
            cpu_usage = random.uniform(10, 90)
            self.add_metric(MonitoringMetric(
                device_id=device.id,
                metric_name="cpu_usage",
                value=cpu_usage,
                unit="%",
                timestamp=timestamp,
                threshold_warning=80.0,
                threshold_critical=95.0
            ))
            
            # 内存利用率
            memory_usage = random.uniform(20, 85)
            self.add_metric(MonitoringMetric(
                device_id=device.id,
                metric_name="memory_usage",
                value=memory_usage,
                unit="%",
                timestamp=timestamp,
                threshold_warning=85.0,
                threshold_critical=95.0
            ))
            
            # 接口流量
            interface_traffic = random.uniform(1000000, 100000000)
            self.add_metric(MonitoringMetric(
                device_id=device.id,
                metric_name="interface_traffic",
                value=interface_traffic,
                unit="bps",
                timestamp=timestamp,
                threshold_warning=80000000.0,
                threshold_critical=95000000.0
            ))
            
        except Exception as e:
            print(f"收集SNMP指标时出错: {e}")
    
    def _check_metric_thresholds(self, metric: MonitoringMetric):
        """检查指标阈值"""
        if metric.threshold_critical and metric.value >= metric.threshold_critical:
            self._generate_alert(
                metric.device_id,
                "threshold_critical",
                "critical",
                f"{metric.metric_name} 达到严重阈值: {metric.value} {metric.unit}"
            )
        elif metric.threshold_warning and metric.value >= metric.threshold_warning:
            self._generate_alert(
                metric.device_id,
                "threshold_warning",
                "warning",
                f"{metric.metric_name} 达到警告阈值: {metric.value} {metric.unit}"
            )
    
    def _generate_status_alert(self, device_id: str, old_status: DeviceStatus, new_status: DeviceStatus):
        """生成状态变化告警"""
        if new_status == DeviceStatus.DOWN:
            severity = "critical"
            message = f"设备从 {old_status.value} 变为 {new_status.value}"
        elif new_status == DeviceStatus.UP and old_status == DeviceStatus.DOWN:
            severity = "info"
            message = f"设备恢复正常: 从 {old_status.value} 变为 {new_status.value}"
        else:
            severity = "warning"
            message = f"设备状态变化: 从 {old_status.value} 变为 {new_status.value}"
        
        self._generate_alert(device_id, "status_change", severity, message)
    
    def _generate_alert(self, device_id: str, alert_type: str, severity: str, message: str):
        """生成告警"""
        alert = {
            'device_id': device_id,
            'alert_type': alert_type,
            'severity': severity,
            'message': message,
            'timestamp': time.time(),
            'acknowledged': False,
            'resolved': False
        }
        
        self.alerts.append(alert)
        
        # 保存到数据库
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO alerts 
            (device_id, alert_type, severity, message, timestamp, acknowledged, resolved)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        ''', (
            device_id, alert_type, severity, message, 
            alert['timestamp'], alert['acknowledged'], alert['resolved']
        ))
        
        conn.commit()
        conn.close()
        
        # 触发事件处理器
        self._trigger_event_handlers('alert_generated', alert)
    
    def get_alerts(self, device_id: str = None, severity: str = None, 
                  unresolved_only: bool = True) -> List[dict]:
        """获取告警"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        query = 'SELECT * FROM alerts WHERE 1=1'
        params = []
        
        if device_id:
            query += ' AND device_id = ?'
            params.append(device_id)
        
        if severity:
            query += ' AND severity = ?'
            params.append(severity)
        
        if unresolved_only:
            query += ' AND resolved = FALSE'
        
        query += ' ORDER BY timestamp DESC'
        
        cursor.execute(query, params)
        
        alerts = []
        for row in cursor.fetchall():
            alerts.append({
                'id': row[0],
                'device_id': row[1],
                'alert_type': row[2],
                'severity': row[3],
                'message': row[4],
                'timestamp': row[5],
                'acknowledged': bool(row[6]),
                'resolved': bool(row[7])
            })
        
        conn.close()
        return alerts
    
    def acknowledge_alert(self, alert_id: int):
        """确认告警"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute(
            'UPDATE alerts SET acknowledged = TRUE WHERE id = ?',
            (alert_id,)
        )
        
        conn.commit()
        conn.close()
    
    def resolve_alert(self, alert_id: int):
        """解决告警"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute(
            'UPDATE alerts SET resolved = TRUE WHERE id = ?',
            (alert_id,)
        )
        
        conn.commit()
        conn.close()
    
    def register_event_handler(self, event_type: str, handler: Callable):
        """注册事件处理器"""
        if event_type not in self.event_handlers:
            self.event_handlers[event_type] = []
        self.event_handlers[event_type].append(handler)
    
    def _trigger_event_handlers(self, event_type: str, data: dict):
        """触发事件处理器"""
        if event_type in self.event_handlers:
            for handler in self.event_handlers[event_type]:
                try:
                    handler(data)
                except Exception as e:
                    print(f"事件处理器执行失败: {e}")
    
    def get_dashboard_data(self) -> dict:
        """获取仪表板数据"""
        total_devices = len(self.devices)
        up_devices = len([d for d in self.devices.values() if d.status == DeviceStatus.UP])
        down_devices = len([d for d in self.devices.values() if d.status == DeviceStatus.DOWN])
        
        critical_alerts = len([a for a in self.get_alerts() if a['severity'] == 'critical'])
        warning_alerts = len([a for a in self.get_alerts() if a['severity'] == 'warning'])
        
        return {
            'devices': {
                'total': total_devices,
                'up': up_devices,
                'down': down_devices,
                'availability': (up_devices / total_devices * 100) if total_devices > 0 else 0
            },
            'alerts': {
                'critical': critical_alerts,
                'warning': warning_alerts,
                'total': critical_alerts + warning_alerts
            }
        }

# 使用示例
async def demo_network_management_platform():
    """网络管理平台演示"""
    platform = NetworkManagementPlatform()
    
    # 添加设备
    devices = [
        NetworkDevice(
            id="router-001",
            name="核心路由器",
            ip_address="192.168.1.1",
            device_type="router",
            location="机房A",
            status=DeviceStatus.UNKNOWN,
            last_seen=time.time()
        ),
        NetworkDevice(
            id="switch-001",
            name="接入交换机",
            ip_address="192.168.1.10",
            device_type="switch",
            location="机房A",
            status=DeviceStatus.UNKNOWN,
            last_seen=time.time()
        ),
        NetworkDevice(
            id="server-001",
            name="Web服务器",
            ip_address="192.168.1.100",
            device_type="server",
            location="机房B",
            status=DeviceStatus.UNKNOWN,
            last_seen=time.time()
        )
    ]
    
    for device in devices:
        platform.add_device(device)
    
    # 注册告警处理器
    def alert_handler(alert_data):
        print(f"[告警] {alert_data['severity'].upper()}: {alert_data['message']}")
    
    platform.register_event_handler('alert_generated', alert_handler)
    
    print("网络管理平台已启动，开始监控设备...")
    
    # 运行一段时间
    await asyncio.sleep(30)
    
    # 获取仪表板数据
    dashboard = platform.get_dashboard_data()
    print(f"\n仪表板数据:")
    print(f"  设备总数: {dashboard['devices']['total']}")
    print(f"  在线设备: {dashboard['devices']['up']}")
    print(f"  离线设备: {dashboard['devices']['down']}")
    print(f"  可用性: {dashboard['devices']['availability']:.1f}%")
    print(f"  严重告警: {dashboard['alerts']['critical']}")
    print(f"  警告告警: {dashboard['alerts']['warning']}")
    
    # 获取设备指标
    for device in devices:
        metrics = platform.get_device_metrics(device.id, hours=1)
        print(f"\n{device.name} 指标 ({len(metrics)} 条):")
        for metric in metrics[-3:]:  # 显示最近3条
            print(f"  {metric.metric_name}: {metric.value} {metric.unit}")
    
    # 获取告警
    alerts = platform.get_alerts()
    print(f"\n当前告警 ({len(alerts)} 条):")
    for alert in alerts[-5:]:  # 显示最近5条
        print(f"  [{alert['severity']}] {alert['message']}")
    
    # 停止所有监控任务
    for device_id in list(platform.monitoring_tasks.keys()):
        platform.stop_device_monitoring(device_id)

if __name__ == "__main__":
    asyncio.run(demo_network_management_platform())

ifTable (1.3.6.1.2.1.2.2)
    |
    +-- ifEntry (1.3.6.1.2.1.2.2.1=x)
            |
            +-- ifIndex(1)        [接口索引号]
            +-- ifDescr(2)        [接口描述]
            +-- ifType(3)         [接口类型]
            +-- ifMtu(4)         [最大传输单元]
            +-- ifSpeed(5)        [接口速率]
            +-- ifPhysAddress(6)  [物理地址]
            +-- ifAdminStatus(7)  [管理状态]
            +-- ifOperStatus(8)   [操作状态]
            +-- ifLastChange(9)   [最后状态改变时间]
            +-- ifInOctets(10)    [输入字节数]
            +-- ifOutOctets(16)   [输出字节数]