diff --git a/libs/psa/constants.py b/libs/psa/constants.py index e660990..e7bbf37 100644 --- a/libs/psa/constants.py +++ b/libs/psa/constants.py @@ -29,4 +29,10 @@ DEFAULT_PRECONDITIONING_PROGRAM = { "program4": {"day": [0, 0, 0, 0, 0, 0, 0], "hour": 34, "minute": 7, "on": 0} } AUTHORIZE_SERVICE = "https://api.mpsa.com/api/connectedcar/v2/oauth/authorize" -REMOTE_URL = "https://api.groupe-psa.com/connectedcar/v4/virtualkey/remoteaccess/token?client_id=" \ No newline at end of file +REMOTE_URL = "https://api.groupe-psa.com/connectedcar/v4/virtualkey/remoteaccess/token?client_id=" +BRAND = {"com.psa.mym.myopel": {"realm": "clientsB2COpel", "brand_code": "OP", "app_name": "MyOpel"}, + "com.psa.mym.mypeugeot": {"realm": "clientsB2CPeugeot", "brand_code": "AP", "app_name": "MyPeugeot"}, + "com.psa.mym.mycitroen": {"realm": "clientsB2CCitroen", "brand_code": "AC", "app_name": "MyCitroen"}, + "com.psa.mym.myds": {"realm": "clientsB2CDS", "brand_code": "DS", "app_name": "MyDS"}, + "com.psa.mym.myvauxhall": {"realm": "clientsB2CVauxhall", "brand_code": "VX", "app_name": "MyVauxhall"} + } \ No newline at end of file diff --git a/libs/psa/setup/apk_parser.py b/libs/psa/setup/apk_parser.py new file mode 100644 index 0000000..9613209 --- /dev/null +++ b/libs/psa/setup/apk_parser.py @@ -0,0 +1,54 @@ +import json +import os + +from androguard.core.bytecodes.apk import APK +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.serialization import pkcs12 + +from libs.psa.constants import BRAND + + +class ApkParser: + def __init__(self, filename, country_code): + self.country_code = country_code + self.filename = filename + self.host_brandid_prod = None + self.site_code = None + self.culture = None + self.client_id = None + self.client_secret = None + + @staticmethod + def __get_cultures_code(file, country_code): + cultures = json.loads(file) + return cultures[country_code]["languages"][0] + + def retrieve_content_from_apk(self): + a = APK(self.filename) + package_name = a.get_package() + resources = a.get_android_resources() # .get_strings_resources() + self.client_id = resources.get_string(package_name, "PSA_API_CLIENT_ID_PROD")[1] + self.client_secret = resources.get_string(package_name, "PSA_API_CLIENT_SECRET_PROD")[1] + self.host_brandid_prod = resources.get_string(package_name, "HOST_BRANDID_PROD")[1] + self.culture = self.__get_cultures_code(a.get_file("res/raw/cultures.json"), self.country_code) + ## Get Customer id + self.site_code = BRAND[package_name]["brand_code"] + "_" + self.country_code + "_ESP" + pfx_cert = a.get_file("assets/MWPMYMA1.pfx") + save_key_to_pem(pfx_cert, b"y5Y2my5B") + + +def save_key_to_pem(pfx_data, pfx_password): + private_key, certificate = pkcs12.load_key_and_certificates(pfx_data, + pfx_password, default_backend())[:2] + try: + os.mkdir("certs") + except FileExistsError: + pass + with open("certs/public.pem", "wb") as f: + f.write(certificate.public_bytes(encoding=serialization.Encoding.PEM)) + + with open("certs/private.pem", "wb") as f: + f.write(private_key.private_bytes(encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.TraditionalOpenSSL, + encryption_algorithm=serialization.NoEncryption())) \ No newline at end of file diff --git a/app_decoder.py b/libs/psa/setup/app_decoder.py similarity index 50% rename from app_decoder.py rename to libs/psa/setup/app_decoder.py index 65e8d83..e2ff7bf 100755 --- a/app_decoder.py +++ b/libs/psa/setup/app_decoder.py @@ -1,84 +1,42 @@ #!/usr/bin/env python3 import json -import os import traceback -from os import path -from androguard.core.bytecodes.apk import APK import requests -from cryptography.hazmat.primitives import serialization -from cryptography.hazmat.primitives.serialization import pkcs12 -from cryptography.hazmat.backends import default_backend from charge_control import ChargeControl, ChargeControls +from libs.psa.constants import BRAND +from libs.psa.setup.apk_parser import ApkParser +from libs.psa.setup.github import urlretrieve_from_github from my_psacc import MyPSACC from mylogger import logger -BRAND = {"com.psa.mym.myopel": {"realm": "clientsB2COpel", "brand_code": "OP", "app_name": "MyOpel"}, - "com.psa.mym.mypeugeot": {"realm": "clientsB2CPeugeot", "brand_code": "AP", "app_name": "MyPeugeot"}, - "com.psa.mym.mycitroen": {"realm": "clientsB2CCitroen", "brand_code": "AC", "app_name": "MyCitroen"}, - "com.psa.mym.myds": {"realm": "clientsB2CDS", "brand_code": "DS", "app_name": "MyDS"}, - "com.psa.mym.myvauxhall": {"realm": "clientsB2CVauxhall", "brand_code": "VX", "app_name": "MyVauxhall"} - } -DOWNLOAD_URL = "https://github.com/flobz/psa_apk/raw/main/" APP_VERSION = "1.33.0" +GITHUB_USER = "flobz" +GITHUB_REPO = "psa_apk" -def save_key_to_pem(pfx_data, pfx_password): - private_key, certificate = pkcs12.load_key_and_certificates(pfx_data, - pfx_password, default_backend())[:2] - try: - os.mkdir("certs") - except FileExistsError: - pass - with open("certs/public.pem", "wb") as f: - f.write(certificate.public_bytes(encoding=serialization.Encoding.PEM)) - - with open("certs/private.pem", "wb") as f: - f.write(private_key.private_bytes(encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.TraditionalOpenSSL, - encryption_algorithm=serialization.NoEncryption())) - - -def urlretrieve(url, path): - with open(path, 'wb') as f: - r = requests.get(url, stream=True) - r.raise_for_status() - for chunk in r.iter_content(1024): - f.write(chunk) - - -def get_cultures_code(file, country_code): - cultures = json.loads(file) - return cultures[country_code]["languages"][0] +def get_content_from_apk(filename: str, country_code: str) -> ApkParser: + apk_parser = ApkParser(filename, country_code) + urlretrieve_from_github(GITHUB_USER, GITHUB_REPO, "", apk_parser.filename) + apk_parser.retrieve_content_from_apk() + return apk_parser def firstLaunchConfig(package_name, client_email, client_password, country_code, # pylint: disable=too-many-locals config_prefix=""): filename = package_name.split(".")[-1] + ".apk" - if not path.exists(filename): - urlretrieve(DOWNLOAD_URL + filename, filename) - a = APK(filename) - package_name = a.get_package() - resources = a.get_android_resources() # .get_strings_resources() - client_id = resources.get_string(package_name, "PSA_API_CLIENT_ID_PROD")[1] - client_secret = resources.get_string(package_name, "PSA_API_CLIENT_SECRET_PROD")[1] - HOST_BRANDID_PROD = resources.get_string(package_name, "HOST_BRANDID_PROD")[1] - REMOTE_REFRESH_TOKEN = None - culture = get_cultures_code(a.get_file("res/raw/cultures.json"), country_code) - ## Get Customer id - site_code = BRAND[package_name]["brand_code"] + "_" + country_code + "_ESP" - pfx_cert = a.get_file("assets/MWPMYMA1.pfx") - save_key_to_pem(pfx_cert, b"y5Y2my5B") + apk_parser = get_content_from_apk(filename, country_code) + try: - res = requests.post(HOST_BRANDID_PROD + "/GetAccessToken", + res = requests.post(apk_parser.host_brandid_prod + "/GetAccessToken", headers={ "Connection": "Keep-Alive", "Content-Type": "application/json", "User-Agent": "okhttp/2.3.0" }, params={"jsonRequest": json.dumps( - {"siteCode": site_code, "culture": "fr-FR", "action": "authenticate", + {"siteCode": apk_parser.site_code, "culture": "fr-FR", "action": "authenticate", "fields": {"USR_EMAIL": {"value": client_email}, "USR_PASSWORD": {"value": client_password}} } @@ -87,7 +45,8 @@ def firstLaunchConfig(package_name, client_email, client_password, country_code, token = res.json()["accessToken"] except Exception as ex: - msg = traceback.format_exc() + f"\nHOST_BRANDID : {HOST_BRANDID_PROD} sitecode: {site_code}" + msg = traceback.format_exc() + f"\nHOST_BRANDID : {apk_parser.host_brandid_prod} " \ + f"sitecode: {apk_parser.site_code}" try: msg += res.text except: # pylint: disable=bare-except @@ -98,11 +57,11 @@ def firstLaunchConfig(package_name, client_email, client_password, country_code, res2 = requests.post( f"https://mw-{BRAND[package_name]['brand_code'].lower()}-m2c.mym.awsmpsa.com/api/v1/user", params={ - "culture": culture, + "culture": apk_parser.culture, "width": 1080, "version": APP_VERSION }, - data=json.dumps({"site_code": site_code, "ticket": token}), + data=json.dumps({"site_code": apk_parser.site_code, "ticket": token}), headers={ "Connection": "Keep-Alive", "Content-Type": "application/json;charset=UTF-8", @@ -125,7 +84,8 @@ def firstLaunchConfig(package_name, client_email, client_password, country_code, logger.error(msg) raise Exception(msg) from ex # Psacc - psacc = MyPSACC(None, client_id, client_secret, REMOTE_REFRESH_TOKEN, customer_id, BRAND[package_name]["realm"], + psacc = MyPSACC(None, apk_parser.client_id, apk_parser.client_secret, + None, customer_id, BRAND[package_name]["realm"], country_code) psacc.connect(client_email, client_password) psacc.save_config(name=config_prefix + "config.json") diff --git a/libs/psa/setup/github.py b/libs/psa/setup/github.py new file mode 100644 index 0000000..9f60c61 --- /dev/null +++ b/libs/psa/setup/github.py @@ -0,0 +1,43 @@ +from hashlib import sha1 + +import requests + +from mylogger import logger + + +def get_github_sha_from_file(user, repo, directory, filename): + res = requests.get("https://api.github.com/repos/{}/{}/git/trees/main:{}".format(user, repo, directory)).json() + file_info = next((file for file in res["tree"] if file['path'] == filename)) + return file_info["sha"] + + +def github_file_need_to_be_downloaded(user, repo, directory, filename): + try: + with open(filename, 'rb') as file_for_hash: + data = file_for_hash.read() + filesize = len(data) + prefix = "blob " + str(filesize) + "\0" + sha_of_downloaded_file = sha1(prefix.encode("utf-8") + data).hexdigest() + sha_of_git_file = get_github_sha_from_file(user, repo, directory, filename) + if sha_of_downloaded_file == sha_of_git_file: + logger.debug("locale file is the latest version") + return False + logger.debug("download last version of file") + except FileNotFoundError: + logger.debug("File not found, download file") + return True + + +def urlretrieve_from_github(user, repo, directory, filename, branch="main"): + if github_file_need_to_be_downloaded(user, repo, directory, filename): + with open(filename, 'wb') as f: + r = requests.get("https://github.com/{}/{}/raw/{}/{}{}".format(user, repo, branch, directory, filename), + headers={ + "Accept": "application/vnd.github.VERSION.raw" + }, + stream=True + ) + + r.raise_for_status() + for chunk in r.iter_content(1024): + f.write(chunk) diff --git a/test/test_unit.py b/test/test_unit.py index 1a0454f..738ea5f 100644 --- a/test/test_unit.py +++ b/test/test_unit.py @@ -7,6 +7,8 @@ from datetime import datetime, timedelta from pytz import UTC import libs.config +from libs.psa.setup.app_decoder import GITHUB_USER, GITHUB_REPO +from libs.psa.setup.github import github_file_need_to_be_downloaded from psa_connectedcar import ApiClient import psa_connectedcar as psacc import reverse_geocode @@ -28,9 +30,8 @@ from web.figures import get_figures, get_battery_curve_fig, get_altitude_fig from deepdiff import DeepDiff - def compare_dict(result, expected): - diff = DeepDiff(expected, result) + diff = DeepDiff(expected, result) if diff != {}: raise AssertionError(str(diff)) return True @@ -95,7 +96,8 @@ class TestUnit(unittest.TestCase): Ecomix.co2_signal_key = key def_country = "FR" Ecomix.get_data_from_co2_signal(latitude, longitude, def_country) - res = Ecomix.get_co2_from_signal_cache(datetime.utcnow().replace(tzinfo=UTC) - timedelta(minutes=5), datetime.now(), def_country) + res = Ecomix.get_co2_from_signal_cache(datetime.utcnow().replace(tzinfo=UTC) - timedelta(minutes=5), + datetime.now(), def_country) assert isinstance(res, float) def test_charge_control(self): @@ -167,7 +169,7 @@ class TestUnit(unittest.TestCase): end_level = 85 Charging.record_charging(car, "InProgress", date0, start_level, latitude, longitude, None, "slow", 20, 60) Charging.record_charging(car, "InProgress", date1, 70, latitude, longitude, "FR", "slow", 20, 60) - Charging.record_charging(car, "InProgress", date1, 70, latitude, longitude, "FR", "slow",20, 60) + Charging.record_charging(car, "InProgress", date1, 70, latitude, longitude, "FR", "slow", 20, 60) Charging.record_charging(car, "InProgress", date2, 80, latitude, longitude, "FR", "slow", 20, 60) Charging.record_charging(car, "Stopped", date3, end_level, latitude, longitude, "FR", "slow", 20, 60) chargings = Charging.get_chargings() @@ -202,7 +204,7 @@ class TestUnit(unittest.TestCase): 'start_at': date0, 'consumption_by_temp': None, 'positions': {'lat': [latitude], - 'long': [longitude]}, + 'long': [longitude]}, 'duration': 40.0, 'speed_average': 28.5, 'distance': 19.0, @@ -231,7 +233,7 @@ class TestUnit(unittest.TestCase): 'start_at': start, 'consumption_by_temp': None, 'positions': {'lat': [latitude], - 'long': [longitude]}, + 'long': [longitude]}, 'duration': 120.0, 'speed_average': 9.5, 'distance': 19.0, @@ -257,6 +259,7 @@ class TestUnit(unittest.TestCase): @rate_limit(2, 10) def test_fct(): pass + test_fct() test_fct() try: @@ -265,6 +268,22 @@ class TestUnit(unittest.TestCase): except RateLimitException: pass + def test_parse_apk(self): + from libs.psa.setup.app_decoder import get_content_from_apk + filename = "mypeugeot.apk" + try: + os.remove(filename) + except FileNotFoundError: + pass + assert get_content_from_apk(filename, "FR") + assert github_file_need_to_be_downloaded(GITHUB_USER, GITHUB_REPO, "", filename) is False + + def test_file_need_to_be_updated(self): + filename = "mypeugeot.apk" + with open(filename, "w") as f: + f.write(" ") + assert github_file_need_to_be_downloaded(GITHUB_USER, GITHUB_REPO, "", filename) is True + if __name__ == '__main__': my_logger(handler_level=os.environ.get("DEBUG_LEVEL", 20)) diff --git a/web/view/config_views.py b/web/view/config_views.py index 087c14d..2fc3d5f 100644 --- a/web/view/config_views.py +++ b/web/view/config_views.py @@ -2,7 +2,7 @@ from dash import callback_context, html, dcc from dash.exceptions import PreventUpdate from flask import request -from app_decoder import firstLaunchConfig +from libs.psa.setup.app_decoder import firstLaunchConfig from libs.config import Config from mylogger import LOG_FILE, logger from otp.otp import new_otp_session