1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
|
#
# This file is part of pysnmp software.
#
# Copyright (c) 2005-2016, Ilya Etingof <ilya@glas.net>
# License: http://pysnmp.sf.net/license.html
#
import random
from pyasn1.type import univ
from pysnmp.proto.secmod.rfc3414.priv import base
from pysnmp.proto.secmod.rfc3414.auth import hmacmd5, hmacsha
from pysnmp.proto.secmod.rfc3414 import localkey
from pysnmp.proto import errind, error
try:
from Crypto.Cipher import AES
except ImportError:
AES = None
random.seed()
# RFC3826
#
class Aes(base.AbstractEncryptionService):
serviceID = (1, 3, 6, 1, 6, 3, 10, 1, 2, 4) # usmAesCfb128Protocol
keySize = 16
_localInt = random.randrange(0, 0xffffffffffffffff)
# 3.1.2.1
def __getEncryptionKey(self, privKey, snmpEngineBoots, snmpEngineTime):
salt = [self._localInt>>56&0xff,
self._localInt>>48&0xff,
self._localInt>>40&0xff,
self._localInt>>32&0xff,
self._localInt>>24&0xff,
self._localInt>>16&0xff,
self._localInt>>8&0xff,
self._localInt&0xff]
if self._localInt == 0xffffffffffffffff:
self._localInt = 0
else:
self._localInt += 1
return self.__getDecryptionKey(privKey, snmpEngineBoots, snmpEngineTime, salt) + (univ.OctetString(salt).asOctets(),)
def __getDecryptionKey(self, privKey, snmpEngineBoots,
snmpEngineTime, salt):
snmpEngineBoots, snmpEngineTime, salt = (
int(snmpEngineBoots), int(snmpEngineTime), salt
)
iv = [snmpEngineBoots>>24&0xff,
snmpEngineBoots>>16&0xff,
snmpEngineBoots>>8&0xff,
snmpEngineBoots&0xff,
snmpEngineTime>>24&0xff,
snmpEngineTime>>16&0xff,
snmpEngineTime>>8&0xff,
snmpEngineTime&0xff] + salt
return privKey[:self.keySize].asOctets(), univ.OctetString(iv).asOctets()
def hashPassphrase(self, authProtocol, privKey):
if authProtocol == hmacmd5.HmacMd5.serviceID:
return localkey.hashPassphraseMD5(privKey)
elif authProtocol == hmacsha.HmacSha.serviceID:
return localkey.hashPassphraseSHA(privKey)
else:
raise error.ProtocolError(
'Unknown auth protocol %s' % (authProtocol,)
)
def localizeKey(self, authProtocol, privKey, snmpEngineID):
if authProtocol == hmacmd5.HmacMd5.serviceID:
localPrivKey = localkey.localizeKeyMD5(privKey, snmpEngineID)
elif authProtocol == hmacsha.HmacSha.serviceID:
localPrivKey = localkey.localizeKeySHA(privKey, snmpEngineID)
else:
raise error.ProtocolError(
'Unknown auth protocol %s' % (authProtocol,)
)
return localPrivKey[:16]
# 3.2.4.1
def encryptData(self, encryptKey, privParameters, dataToEncrypt):
if AES is None:
raise error.StatusInformation(
errorIndication=errind.encryptionError
)
snmpEngineBoots, snmpEngineTime, salt = privParameters
# 3.3.1.1
aesKey, iv, salt = self.__getEncryptionKey(
encryptKey, snmpEngineBoots, snmpEngineTime
)
# 3.3.1.3
aesObj = AES.new(aesKey, AES.MODE_CFB, iv, segment_size=128)
# PyCrypto seems to require padding
dataToEncrypt = dataToEncrypt + univ.OctetString((0,) * (16-len(dataToEncrypt)%16)).asOctets()
ciphertext = aesObj.encrypt(dataToEncrypt)
# 3.3.1.4
return univ.OctetString(ciphertext), univ.OctetString(salt)
# 3.2.4.2
def decryptData(self, decryptKey, privParameters, encryptedData):
if AES is None:
raise error.StatusInformation(
errorIndication=errind.decryptionError
)
snmpEngineBoots, snmpEngineTime, salt = privParameters
# 3.3.2.1
if len(salt) != 8:
raise error.StatusInformation(
errorIndication=errind.decryptionError
)
# 3.3.2.3
aesKey, iv = self.__getDecryptionKey(
decryptKey, snmpEngineBoots, snmpEngineTime, salt
)
aesObj = AES.new(aesKey, AES.MODE_CFB, iv, segment_size=128)
# PyCrypto seems to require padding
encryptedData = encryptedData + univ.OctetString((0,) * (16-len(encryptedData)%16)).asOctets()
# 3.3.2.4-6
return aesObj.decrypt(encryptedData.asOctets())
|
