从Java到python的API调用移植(Kostal Plenticore Inverter) [英] API call portation from java to python (Kostal Plenticore Inverter)

问题描述

我试图通过api从逆变器中读取传感器数据. (科斯塔尔PLENTICORE plus) 由于Kostal缺乏文档，因此无法正常工作.身份验证是这里的大问题.但是我只是从Openhab找到了代码.

Im trying to read sensor data out of my inverter via the api. (Kostal PLENTICORE plus) Due to the lack of documentation by Kostal I haven't got it working. The authentication is the big problem here. But I just found the code from Openhab.
ThirdGenerationEncryptionHelper ThirdGenerationHandler
And now im trying to port it as easy as possible to python.
My Code for now:

    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吗? 除此之外，我在加密或这种身份验证方面经验不足.有人可以告诉我我在做的员工是否正确?

Now to my questions: I stuck with the creation of the client proof (function createClientProof). Can someone help me to do the XOR like its done in java ? Besides that, i have not much experience with encryption or this kind of authentification. Can someone tell me if that staff im doing is correct ?

原始的:

/**
 * 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);
}

感谢您的帮助

推荐答案

我刚刚添加了身份验证的其他步骤，因此该代码完全覆盖了auth-progress.我在逆变器上使用的是SW版本01.13.04122和API版本0.2.0.请检查您的版本.

I just added the further steps of authentication so the auth-progress is completely coverd with that code. Im using SW-Version 01.13.04122 and API-Version 0.2.0 on the inverter. Please check your version.

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

这篇关于从Java到python的API调用移植(Kostal Plenticore Inverter)的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！