从 java 到 python 的 API 调用移植（Kostal Plenticore Inverter）

Question

我试图通过 api 从我的逆变器中读取传感器数据。 (Kostal PLENTICORE plus) 由于 Kostal 缺乏文档，我没有让它工作。身份验证是这里的大问题。但我刚刚从 Openhab 找到了代码。
ThirdGenerationEncryptionHelper ThirdGenerationHandler
现在我试图将它尽可能简单地移植到 python。
我现在的代码：

    import requests
    import random
    import string
    import json
    import hashlib
    import hmac
    import hashlib
    import binascii

    def randomString(stringLength=10):
        """Generate a random string of fixed length """
        letters = string.ascii_lowercase
        return ''.join(random.choice(letters) for i in range(stringLength))

    def getPBKDF2Hash(password, salt, rounds):
        key = hashlib.pbkdf2_hmac(
        'sha256', # The hash digest algorithm for HMAC
        password.encode('utf-8'), # Convert the password to bytes
        salt, # Provide the salt
        rounds # It is recommended to use at least 100,000 iterations of SHA-256
        )
        return key

    def create_sha256_signature(byte_key, message):
        #byte_key = binascii.unhexlify(key)
        message = message.encode()
        return hmac.new(byte_key, message, hashlib.sha256).hexdigest().upper()

    def createClientProof(clientSignature, serverSignature):
            clientlength = len(clientSignature.encode('utf-8'))
            result = []
            #for i in range(clientlength):
            #    result[i] = (0xff & (bytes(clientSignature[i]) ^ bytes(serverSignature[i])))
            return result**

    username="user"
    password= "A123456789"
    url = 'http://192.168.1.23/api/v1/'
    clientNonce = randomString(16)
    reqstart = {"username": username, "nonce": clientNonce}

    a = requests.post(url+'auth/start', json=reqstart)
    anserstart = json.loads(a.text)

    serverNonce = anserstart['nonce']
    transactionId = anserstart['transactionId']
    salt = anserstart['salt']
    rounds = anserstart['rounds']

    saltedpassword = getPBKDF2Hash(password, salt, rounds)
    clientkey = create_sha256_signature(saltedpassword, "Client Key")
    serverkey = create_sha256_signature(saltedpassword, "Server Key")
    storedKey = hashlib.sha256(clientkey).hexdigest()
    authMessage = "n={},r={},r={},s={},i={},c=biw,r={}"
    authMessage.format(username, clientNonce, serverNonce, salt, rounds, serverNonce)
    clientSignature = create_sha256_signature(storedKey, authMessage)
    serverSignature = create_sha256_signature(storedKey, serverkey)

    print(anserstart)
    #print(saltedpassword)
    #print(clientkey)
    #print(serverkey)
    #print(storedKey)
    print(clientSignature)
    print(serverSignature)
    print(createClientProof(clientSignature,serverSignature))
    #reqfinish = {"proof": "", "transactionId": transactionId}

    #b = requests.post(url+'auth/start', json=reqfinish)
    #answerfinish = json.loads(b.text)
    #print(answerfinish)

现在我的问题： 我坚持创建客户证明（函数 createClientProof）。有人可以帮我像在 java 中那样做 XOR 吗？ 除此之外，我在加密或这种身份验证方面没有太多经验。谁能告诉我我做的那个员工是否正确？

原件：

/**
 * This method generates the HMACSha256 encrypted value of the given value
 *
 * @param password       Password used for encryption
 * @param valueToEncrypt value to encrypt
 * @return encrypted value
 * @throws InvalidKeyException      thrown if the key generated from the password is invalid
 * @throws NoSuchAlgorithmException thrown if HMAC SHA 256 is not supported
 */
static byte[] getHMACSha256(byte[] password, String valueToEncrypt)
        throws InvalidKeyException, NoSuchAlgorithmException {
    SecretKeySpec signingKey = new SecretKeySpec(password, HMAC_SHA256_ALGORITHM);
    Mac mac = Mac.getInstance(HMAC_SHA256_ALGORITHM);
    mac.init(signingKey);
    mac.update(valueToEncrypt.getBytes());
    return mac.doFinal();
}

/**
 * This methods generates the client proof.
 * It is calculated as XOR between the {@link clientSignature} and the {@link serverSignature}
 *
 * @param clientSignature client signature
 * @param serverSignature server signature
 * @return client proof
 */
static String createClientProof(byte[] clientSignature, byte[] serverSignature) {
    byte[] result = new byte[clientSignature.length];
    for (int i = 0; i < clientSignature.length; i++) {
        result[i] = (byte) (0xff & (clientSignature[i] ^ serverSignature[i]));
    }
    return Base64.getEncoder().encodeToString(result);
}

/**
 * Create the PBKDF2 hash
 *
 * @param password password
 * @param salt     salt
 * @param rounds   rounds
 * @return hash
 * @throws NoSuchAlgorithmException if PBKDF2WithHmacSHA256 is not supported
 * @throws InvalidKeySpecException  if the key specification is not supported
 */
static byte[] getPBKDF2Hash(String password, byte[] salt, int rounds)
        throws NoSuchAlgorithmException, InvalidKeySpecException {
    PBEKeySpec spec = new PBEKeySpec(password.toCharArray(), salt, rounds, 256);
    SecretKeyFactory skf = SecretKeyFactory.getInstance("PBKDF2WithHmacSHA256");
    return skf.generateSecret(spec).getEncoded();
}

/**
 * Create the SHA256 hash value for the given byte array
 *
 * @param valueToHash byte array to get the hash value for
 * @return the hash value
 * @throws NoSuchAlgorithmException if SHA256 is not supported
 */
static byte[] getSha256Hash(byte[] valueToHash) throws NoSuchAlgorithmException {
    return MessageDigest.getInstance(SHA_256_HASH).digest(valueToHash);
}

感谢您的帮助

Answer 1

我刚刚写了一些python代码来连接逆变器。到目前为止，您可以看到如何处理 XOR 操作。 现在 /auth/start 和 /auth/finish 完成了。下一步是 /auth/create_session ，它也需要一些加密操作。如果你在这方面工作，如果你能发布你的结果会很好。

import random
import string
import base64
import json
import requests
import hashlib
import os
import hmac

USER_TYPE = "user"
PASSWD = 'yourSecretPassword'
BASE_URL = "http://xxx.xxx.xxx.xxx/api/v1"
AUTH_START = "/auth/start"
AUTH_FINISH = "/auth/finish"
AUTH_CREATE_SESSION = "/auth/create_session"

def randomString(stringLength):
    letters = string.ascii_letters
    return ''.join(random.choice(letters) for i in range(stringLength))

u = randomString(12)
u = base64.b64encode(u.encode('utf-8')).decode('utf-8')

step1 = {
  "username": USER_TYPE,
  "nonce": u
}
step1 = json.dumps(step1)

url = BASE_URL + AUTH_START
headers = {'Content-type': 'application/json', 'Accept': 'application/json'}
response = requests.post(url, data=step1, headers=headers)
response = json.loads(response.text)
i = response['nonce']
e = response['transactionId']
o = response['rounds']
a = response['salt']
bitSalt = base64.b64decode(a)

def getPBKDF2Hash(password, bytedSalt, rounds):
    return hashlib.pbkdf2_hmac('sha256', password.encode('utf-8'), bytedSalt, rounds)

r = getPBKDF2Hash(PASSWD,bitSalt,o)
s = hmac.new(r, "Client Key".encode('utf-8'), hashlib.sha256).digest()
c = hmac.new(r, "Server Key".encode('utf-8'), hashlib.sha256).digest()
_ = hashlib.sha256(s).digest()
d = "n=user,r="+u+",r="+i+",s="+a+",i="+str(o)+",c=biws,r="+i
g = hmac.new(_, d.encode('utf-8'), hashlib.sha256).digest()
p = hmac.new(c, d.encode('utf-8'), hashlib.sha256).digest()
f = bytes(a ^ b for (a, b) in zip(s, g))
proof = base64.b64encode(f).decode('utf-8')

step2 = {
  "transactionId": e,
  "proof": proof
}
step2 = json.dumps(step2)

url = BASE_URL + AUTH_FINISH
headers = {'Content-type': 'application/json', 'Accept': 'application/json'}
response = requests.post(url, data=step2, headers=headers)
response = json.loads(response.text)
token = response['token']
signature = response['signature']

# TODO more encryption stuff 

Answer 2

我刚刚添加了进一步的身份验证步骤，因此该代码完全涵盖了身份验证进度。我在逆变器上使用 SW-Version 01.13.04122 和 API-Version 0.2.0。请检查您的版本。

import sys
import random
import string
import base64
import json
import requests
import hashlib
import os
import hmac
from Crypto.Cipher import AES
import binascii
# pip install pycryptodome

USER_TYPE = "user"
PASSWD = 'yourSecretPassword'
BASE_URL = "http://xxx.xxx.xxx.xxx/api/v1"
AUTH_START = "/auth/start"
AUTH_FINISH = "/auth/finish"
AUTH_CREATE_SESSION = "/auth/create_session"
ME = "/auth/me"

def randomString(stringLength):
    letters = string.ascii_letters
    return ''.join(random.choice(letters) for i in range(stringLength))

u = randomString(12)
u = base64.b64encode(u.encode('utf-8')).decode('utf-8')

step1 = {
  "username": USER_TYPE,
  "nonce": u
}
step1 = json.dumps(step1)

url = BASE_URL + AUTH_START
headers = {'Content-type': 'application/json', 'Accept': 'application/json'}
response = requests.post(url, data=step1, headers=headers)
response = json.loads(response.text)
i = response['nonce']
e = response['transactionId']
o = response['rounds']
a = response['salt']
bitSalt = base64.b64decode(a)

def getPBKDF2Hash(password, bytedSalt, rounds):
    return hashlib.pbkdf2_hmac('sha256', password.encode('utf-8'), bytedSalt, rounds)

r = getPBKDF2Hash(PASSWD,bitSalt,o)
s = hmac.new(r, "Client Key".encode('utf-8'), hashlib.sha256).digest()
c = hmac.new(r, "Server Key".encode('utf-8'), hashlib.sha256).digest()
_ = hashlib.sha256(s).digest()
d = "n=user,r="+u+",r="+i+",s="+a+",i="+str(o)+",c=biws,r="+i
g = hmac.new(_, d.encode('utf-8'), hashlib.sha256).digest()
p = hmac.new(c, d.encode('utf-8'), hashlib.sha256).digest()
f = bytes(a ^ b for (a, b) in zip(s, g))
proof = base64.b64encode(f).decode('utf-8')

step2 = {
  "transactionId": e,
  "proof": proof
}
step2 = json.dumps(step2)

url = BASE_URL + AUTH_FINISH
headers = {'Content-type': 'application/json', 'Accept': 'application/json'}
response = requests.post(url, data=step2, headers=headers)
response = json.loads(response.text)
token = response['token']
signature = response['signature']

y = hmac.new(_, "Session Key".encode('utf-8'), hashlib.sha256)
y.update(d.encode('utf-8'))
y.update(s)
P = y.digest()
protocol_key = P
t = os.urandom(16)

e2 = AES.new(protocol_key,AES.MODE_GCM,t)
e2, authtag = e2.encrypt_and_digest(token.encode('utf-8'))

step3 = {
  "transactionId": e,
  "iv": base64.b64encode(t).decode('utf-8'),
  "tag": base64.b64encode(authtag).decode("utf-8"),
  "payload": base64.b64encode(e2).decode('utf-8')
}
step3 = json.dumps(step3)

headers = { 'Content-type': 'application/json', 'Accept': 'application/json' }
url = BASE_URL + AUTH_CREATE_SESSION
response = requests.post(url, data=step3, headers=headers)
response = json.loads(response.text)
sessionId = response['sessionId']

#create a new header with the new Session-ID for all further requests
headers = { 'Content-type': 'application/json', 'Accept': 'application/json', 'authorization': "Session " + sessionId }
url = BASE_URL + ME
response = requests.get(url = url, headers = headers)
response = json.loads(response.text)
authOK = response['authenticated']
if not authOK:
    print("authorization NOT OK")
    sys.exit()

url = BASE_URL + "/info/version"
response = requests.get(url = url, headers = headers)
response = json.loads(response.text)
swversion = response['sw_version']
apiversion = response['api_version']
hostname = response['hostname']
name = response['name']
print("Connected to the inverter " + name + "/" + hostname + " with SW-Version " + swversion + " and API-Version " + apiversion)

# Auth OK, now send your desired requests