github/psa_car_controller
1
0
Fork 0
mirror of https://github.com/flobz/psa_car_controller.git synced 2026-08-26 10:17:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

check if apk changed

Browse Source
This commit is contained in:
Florian Bezannier
2021-12-01 20:41:28 +01:00
parent 21c2d7754f
commit acddabdf47
6 changed files with 150 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/psa/constants.py
+7 -1
View File
@@ -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="
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"}
}
libs/psa/setup/apk_parser.py
+54
View File
@@ -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()))
app_decoder.py → libs/psa/setup/app_decoder.py
+20 -60
View File
@@ -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")
libs/psa/setup/github.py
+43
View File
@@ -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)
test/test_unit.py
+25 -6
View File
@@ -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))
web/view/config_views.py
+1 -1
View File
@@ -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