Adding basic OpenPAYGO Token support - Encoding

Author: benmod

LICENSE

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2023 OpenPAYGO
+Copyright (c) 2023 Eternum Ltd T/A Solaris Offgrid
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

README.md

@@ -1,2 +1,69 @@
-# openpaygo-python
-A Python library for adding support for OpenPAYGO technologies on servers. 
+# OpenPAYGOToken - Python Lib
+
+This repository contains the Python library for using implementing the different OpenPAYGOToken Suite technologies on your server (for generating tokens and decoding openpaygo metrics payloads) or device (for decoding tokens and making openpaygo metrics payloads).  
+
+<p align="center">
+  <img
+    alt="Project Status"
+    src="https://img.shields.io/badge/Project%20Status-beta-orange"
+  >
+  <a href="https://github.com/openpaygo/metrics/blob/main/LICENSE" target="_blank">
+    <img
+      alt="License"
+      src="https://img.shields.io/github/license/openpaygo/openpaygo-python"
+    >
+  </a>
+</p>
+
+This open-source project was sponsored by: 
+- Solaris Offgrid
+- EnAccess
+
+## Table of Contents
+
+  - [Installing the library](#installing-the-library)
+  - [Getting Started - Generating Token](#getting-started---generating-token)
+
+
+## Installing the library
+
+You can install the library by running `pip install openpaygo` or adding `openpaygo` in your requirements.txt file and running `pip install -r requirements.txt`. 
+
+## Getting Started - Generating Token
+
+You can use the `generate_token()` function to generate an OpenPAYGOToken Token. The function takes the following parameters, and they should match the configuration in the hardware of the device: 
+
+- `secret_key` (required): The secret key of the device
+- `value` (required): The value to be passed in the token (typically the number of days of activation)
+- `count` (required): The token count used to make the last token. 
+- `set_mode` (optional): If set to `true`, the generated token will be a Set Time token, otherwise it is Add Time token. 
+- `starting_code` (optional): If not provided, it is generated according to the method defined in the standard (SipHash-2-4 of the key, transformed to digit by the same method as the token generation).
+- `value_divider` (optional): The dividing factor used for the value. 
+- `restricted_digit_set` (optional): If set to `true`, the the restricted digit set will be used (only digits from 1 to 4). 
+- `extended_token` (optional): If set to `true` then a larger token will be generated, able to contain values up to 999999. This is for special use cases of each device, such as settings change, and is not set in the standard. 
+
+
+The function returns the `token` as a string as well as the `updated_count` as a number. 
+
+
+**Example:**
+
+```
+from openpaygo.token import generate_token
+from myexampleproject import device_getter
+
+# We get a device with the parameters we need from our database, this will be specific to your project
+device = device_getter(serial=1234)
+
+# We get the new token and update the count
+new_token, device.count = generate_token(
+  secret_key=device.secret_key,
+  value=7,
+  count=device.count
+)
+
+print('Token: '+new_token)
+device.save() # We save the new count that we set for the device
+```
+
+## Getting Started - Decoding a Token

requirements.txt

@@ -0,0 +1 @@
+siphash==0.0.1

setup.cfg

@@ -0,0 +1,14 @@
+[metadata]
+name = openpaygo
+version = 0.1.0
+url = https://github.com/openpaygo/openpaygo-python
+description-file=README.md
+license_files=LICENSE
+
+[options]
+python_requires = >=2.7
+packages = find:
+include_package_data = true
+zip_safe = false
+install_requires =
+    siphash >= 0.0.1

setup.py

@@ -0,0 +1,14 @@
+from setuptools import setup, find_packages
+
+
+setup(
+    name="openpaygo", 
+    packages=find_packages(),
+    version='0.1.0',
+    license='MIT',
+    author="Solaris Offgrid",
+    url='https://github.com/openpaygo/openpaygo-python',
+    install_requires=[
+        'siphash',
+    ],
+)

token/__init__.py

@@ -0,0 +1,104 @@
+from shared import OpenPAYGOTokenShared
+from shared_extended import OpenPAYGOTokenSharedExtended
+
+
+class OpenPAYGOTokenDecoder(object):
+    MAX_TOKEN_JUMP = 64
+    MAX_TOKEN_JUMP_COUNTER_SYNC = 100
+    MAX_UNUSED_OLDER_TOKENS = 8*2
+
+    @classmethod
+    def get_activation_value_count_and_type_from_token(cls, token, starting_code, key, last_count,
+                                                       restricted_digit_set=False, used_counts=None):
+        if restricted_digit_set:
+            token = OpenPAYGOTokenShared.convert_from_4_digit_token(token)
+        valid_older_token = False
+        token_base = OpenPAYGOTokenShared.get_token_base(token)  # We get the base of the token
+        current_code = OpenPAYGOTokenShared.put_base_in_token(starting_code, token_base)  # We put it into the starting code
+        starting_code_base = OpenPAYGOTokenShared.get_token_base(starting_code)  # We get the base of the starting code
+        value = cls._decode_base(starting_code_base, token_base)  # If there is a match we get the value from the token
+        # We try all combination up until last_count + TOKEN_JUMP, or to the larger jump if syncing counter
+        # We could start directly the loop at the last count if we kept the token value for the last count
+        if value == OpenPAYGOTokenShared.COUNTER_SYNC_VALUE:
+            max_count_try = last_count + cls.MAX_TOKEN_JUMP_COUNTER_SYNC + 1
+        else:
+            max_count_try = last_count + cls.MAX_TOKEN_JUMP + 1
+        for count in range(0, max_count_try):
+            masked_token = OpenPAYGOTokenShared.put_base_in_token(current_code, token_base)
+            if count % 2:
+                this_type = OpenPAYGOTokenShared.TOKEN_TYPE_SET_TIME
+            else:
+                this_type = OpenPAYGOTokenShared.TOKEN_TYPE_ADD_TIME
+            if masked_token == token:
+                if cls._count_is_valid(count, last_count, value, this_type, used_counts):
+                    return value, count, this_type
+                else:
+                    valid_older_token = True
+            current_code = OpenPAYGOTokenShared.generate_next_token(current_code, key)  # If not we go to the next token
+        if valid_older_token:
+            return -2, None, None
+        return None, None, None
+
+    @classmethod
+    def _count_is_valid(cls, count, last_count, value, type, used_counts):
+        if value == OpenPAYGOTokenShared.COUNTER_SYNC_VALUE:
+            if count > last_count - 30:
+                return True
+        elif count > last_count:
+            return True
+        elif cls.MAX_UNUSED_OLDER_TOKENS > 0:
+            if count > last_count - cls.MAX_UNUSED_OLDER_TOKENS:
+                if count not in used_counts and type == OpenPAYGOTokenShared.TOKEN_TYPE_ADD_TIME:
+                    return True
+        return False
+
+    @classmethod
+    def update_used_counts(cls, past_used_counts, value, new_count, type):
+        highest_count = max(past_used_counts) if past_used_counts else 0
+        if new_count > highest_count:
+            highest_count = new_count
+        bottom_range = highest_count-cls.MAX_UNUSED_OLDER_TOKENS
+        used_counts = []
+        if type != OpenPAYGOTokenShared.TOKEN_TYPE_ADD_TIME or value == OpenPAYGOTokenShared.COUNTER_SYNC_VALUE or value == OpenPAYGOTokenShared.PAYG_DISABLE_VALUE:
+            # If it is not an Add-Time token, we mark all the past tokens as used in the range
+            for count in range(bottom_range, highest_count+1):
+                used_counts.append(count)
+        else:
+            # If it is an Add-Time token, we just mark the tokens actually used in the range
+            for count in range(bottom_range, highest_count+1):
+                if count == new_count or count in past_used_counts:
+                    used_counts.append(count)
+        return used_counts
+
+    @classmethod
+    def _decode_base(cls, starting_code_base, token_base):
+        decoded_value = token_base - starting_code_base
+        if decoded_value < 0:
+            return decoded_value + 1000
+        else:
+            return decoded_value
+
+    @classmethod
+    def get_activation_value_count_from_extended_token(cls, token, starting_code, key, last_count,
+                                                       restricted_digit_set=False):
+        if restricted_digit_set:
+            token = OpenPAYGOTokenSharedExtended.convert_from_4_digit_token(token)
+        token_base = OpenPAYGOTokenSharedExtended.get_token_base(token) # We get the base of the token
+        current_code = OpenPAYGOTokenSharedExtended.put_base_in_token(starting_code, token_base) # We put it into the starting code
+        starting_code_base = OpenPAYGOTokenSharedExtended.get_token_base(starting_code) # We get the base of the starting code
+        value = cls._decode_base_extended(starting_code_base, token_base)  # If there is a match we get the value from the token
+        for count in range(0, 30):
+            masked_token = OpenPAYGOTokenSharedExtended.put_base_in_token(current_code, token_base)
+            if masked_token == token and count > last_count:
+                clean_count = count-1
+                return value, clean_count
+            current_code = OpenPAYGOTokenSharedExtended.generate_next_token(current_code, key) # If not we go to the next token
+        return None, None, None
+
+    @classmethod
+    def _decode_base_extended(cls, starting_code_base, token_base):
+        decoded_value = token_base - starting_code_base
+        if decoded_value < 0:
+            return decoded_value + 1000000
+        else:
+            return decoded_value

token/decode_token.py

@@ -0,0 +1,89 @@
+from shared import OpenPAYGOTokenShared
+from shared_extended import OpenPAYGOTokenSharedExtended
+
+
+class OpenPAYGOTokenEncoder(object):
+
+    @classmethod
+    def generate_token(cls, secret_key, value, count, set_mode=False, starting_code=None, value_divider=1, restricted_digit_set=False, extended_token=False):
+        if not starting_code:
+            # We generate the starting code from the key if not provided
+            starting_code = cls.generate_starting_code(secret_key)
+        if set_mode:
+            mode = OpenPAYGOTokenShared.TOKEN_TYPE_SET_TIME
+        else:
+            mode = OpenPAYGOTokenShared.TOKEN_TYPE_ADD_TIME
+        value = int(round(value / value_divider))
+        if extended_token:
+            return cls.generate_extended_token(starting_code, secret_key, value, count, mode, restricted_digit_set)
+        else:
+            return cls.generate_standard_token(starting_code, secret_key, value, count, mode, restricted_digit_set)
+
+    @classmethod
+    def generate_standard_token(cls, starting_code=None, key=None, value=None, count=None,
+                                mode=OpenPAYGOTokenShared.TOKEN_TYPE_SET_TIME, restricted_digit_set=False):
+        # We get the first 3 digits with encoded value
+        starting_code_base = OpenPAYGOTokenShared.get_token_base(starting_code)
+        token_base = cls._encode_base(starting_code_base, value)
+        current_token = OpenPAYGOTokenShared.put_base_in_token(starting_code, token_base)
+        new_count = cls._get_new_count(count, mode)
+        for xn in range(0, new_count):
+            current_token = OpenPAYGOTokenShared.generate_next_token(current_token, key)
+        final_token = OpenPAYGOTokenShared.put_base_in_token(current_token, token_base)
+        if restricted_digit_set:
+            final_token = OpenPAYGOTokenShared.convert_to_4_digit_token(final_token)
+            final_token = '{:015d}'.format(final_token)
+        else:
+            final_token = '{:09d}'.format(final_token)
+        return new_count, final_token
+
+    @classmethod
+    def _encode_base(cls, base, number):
+        if number + base > 999:
+            return number + base - 1000
+        else:
+            return number + base
+
+    @classmethod
+    def generate_extended_token(cls, starting_code, key, value, count, mode=OpenPAYGOTokenShared.TOKEN_TYPE_SET_TIME, restricted_digit_set=False):
+        starting_code_base = OpenPAYGOTokenSharedExtended.get_token_base(starting_code)
+        token_base = cls._encode_base_extended(starting_code_base, value)
+        current_token = OpenPAYGOTokenSharedExtended.put_base_in_token(starting_code, token_base)
+        new_count = cls._get_new_count(count, mode)
+        for xn in range(0, new_count):
+            current_token = OpenPAYGOTokenSharedExtended.generate_next_token(current_token, key)
+        final_token = OpenPAYGOTokenSharedExtended.put_base_in_token(current_token, token_base)
+        if restricted_digit_set:
+            final_token = OpenPAYGOTokenSharedExtended.convert_to_4_digit_token(final_token)
+            final_token = '{:020d}'.format(final_token)
+        else:
+            final_token = '{:012d}'.format(final_token)
+        return new_count, final_token
+    
+    @classmethod
+    def generate_starting_code(cls, key):
+        # We make a hash of the key
+        starting_hash = OpenPAYGOTokenShared.generate_hash(key, key)
+        return OpenPAYGOTokenShared.convert_hash_to_token(starting_hash)
+
+    @classmethod
+    def _encode_base_extended(cls, base, number):
+        if number + base > 999999:
+            return number + base - 1000000
+        else:
+            return number + base
+        
+    @classmethod
+    def _get_new_count(cls, count, mode):
+        current_count_odd = count % 2
+        if mode == OpenPAYGOTokenShared.TOKEN_TYPE_SET_TIME:
+            if current_count_odd: # Odd numbers are for Set Time
+                new_count = count+2
+            else:
+                new_count = count+1
+        else:
+            if current_count_odd: # Even numbers are for Add Time
+                new_count = count+1
+            else:
+                new_count = count+2
+        return new_count

token/encode_token.py

@@ -0,0 +1,83 @@
+import siphash
+import struct
+
+
+class OpenPAYGOTokenShared(object):
+    MAX_BASE = 999
+    MAX_ACTIVATION_VALUE = 995
+    PAYG_DISABLE_VALUE = 998
+    COUNTER_SYNC_VALUE = 999
+    TOKEN_VALUE_OFFSET = 1000
+    TOKEN_TYPE_SET_TIME = 1
+    TOKEN_TYPE_ADD_TIME = 2
+
+    @classmethod
+    def get_token_base(cls, code):
+        return int(code) % cls.TOKEN_VALUE_OFFSET
+
+    @classmethod
+    def put_base_in_token(cls, token, token_base):
+        if token_base > cls.MAX_BASE:
+            Exception('INVALID_VALUE')
+        return token - cls.get_token_base(token) + token_base
+
+    @classmethod
+    def generate_next_token(cls, last_code, key):
+        conformed_token = struct.pack('>L', last_code) # We convert the token to bytes
+        conformed_token += conformed_token # We duplicate it to fit the minimum length
+        token_hash = cls.generate_hash(key, conformed_token) # We hash it
+        new_token = cls.convert_hash_to_token(token_hash) # We convert to token and return
+        return new_token
+
+    @classmethod
+    def convert_hash_to_token(cls, this_hash):
+        hash_int = struct.pack('>Q', this_hash) # We convert the hash to bytes
+        hi_hash = struct.unpack('>L', hash_int[0:4])[0]  # We split it in two 32bits INT
+        lo_hash = struct.unpack('>L', hash_int[4:8])[0]
+        result_hash = hi_hash ^ lo_hash # We XOR the two together to get a single 32bits INT
+        token = cls._convert_to_29_5_bits(result_hash) # We convert the 32bits value to an INT no greater than 9 digits
+        return token
+
+    @classmethod
+    def _convert_to_29_5_bits(cls, source):
+        mask = ((1 << (32 - 2 + 1)) - 1) << 2
+        temp = (source & mask) >> 2
+        if temp > 999999999:
+            temp = temp - 73741825
+        return temp
+
+    @classmethod
+    def convert_to_4_digit_token(cls, source):
+        restricted_digit_token = ''
+        bit_array = cls._bit_array_from_int(source, 30)
+        for i in range(15):
+            this_array = bit_array[i*2:(i*2)+2]
+            restricted_digit_token += str(cls._bit_array_to_int(this_array)+1)
+        return int(restricted_digit_token)
+
+    @classmethod
+    def convert_from_4_digit_token(cls, source):
+        bit_array = []
+        for digit in str(source):
+            digit = int(digit) - 1
+            this_array = cls._bit_array_from_int(digit, 2)
+            bit_array += this_array
+        return cls._bit_array_to_int(bit_array)
+    
+    @classmethod
+    def generate_hash(cls, key, value):
+        return siphash.SipHash_2_4(key, value).hash()
+
+    @classmethod
+    def _bit_array_to_int(cls, bit_array):
+        integer = 0
+        for bit in bit_array:
+            integer = (integer << 1) | bit
+        return integer
+
+    @classmethod
+    def _bit_array_from_int(cls, source, bits):
+        bit_array = []
+        for i in range(bits):
+            bit_array += [bool(source & (1 << (bits - 1 - i)))]
+        return bit_array