diff --git a/arielle.py b/arielle.py new file mode 100644 index 0000000..882d02e --- /dev/null +++ b/arielle.py @@ -0,0 +1,44 @@ +import time +import pprint + +from we_connect import wci +from credentials import user, password +from we_connect.domains import Domain + + +def main(): + w = wci.WeConnectId(user, password) + + vehicles = w.get('/vehicles') + print('list of vehicles:') + for key in vehicles: + print(key) + + pprint.pprint(vehicles) + domains = [Domain.ALL_CAPABLE] + endpoint = '/vehicles/' + vehicles['data'][0]['vin'] + '/selectivestatus?jobs=' + count = len(domains) + for domain in domains: + endpoint += domain.value + count -= 1 + if count > 0: + endpoint += ',' + +# endpoint = '/vehicles/' + vehicles['data'][0]['vin'] + f'/selectivestatus?jobs=access,charging,climatisation,climatisationTimers,fuelStatus,userCapabilities,vehicleLights,vehicleHealthInspection,oilLevel,measurements' + print(f'Endpoint: {endpoint}') + while True: + start = time.time() + try: + data = w.get(endpoint) + print(data) + + except Exception: + pass + + stop = time.time() + diff = stop - start + time.sleep(60-diff) + + +if __name__ == '__main__': + main() diff --git a/credentials.py b/credentials.py new file mode 100644 index 0000000..4fd8f7b --- /dev/null +++ b/credentials.py @@ -0,0 +1,5 @@ +#user = 'jens.ahrensfeld@gmx.de' +#password = '2SaveMyCar' + +user = 'alex.thoni@gmx.de' +password = 'Arielle2024!' diff --git a/wci/domains.py b/wci/domains.py new file mode 100644 index 0000000..a47cf5d --- /dev/null +++ b/wci/domains.py @@ -0,0 +1,27 @@ +import enum + +class Domain(enum.Enum): + ALL = 'all' + ALL_CAPABLE = 'allCapable' + ACCESS = 'access' + ACTIVEVENTILATION = 'activeventilation' + AUTOMATION = 'automation' + AUXILIARY_HEATING = 'auxiliaryheating' + USER_CAPABILITIES = 'userCapabilities' + CHARGING = 'charging' + CHARGING_PROFILES = 'chargingProfiles' + BATTERY_CHARGING_CARE = 'batteryChargingCare' + CLIMATISATION = 'climatisation' + CLIMATISATION_TIMERS = 'climatisationTimers' + DEPARTURE_TIMERS = 'departureTimers' + FUEL_STATUS = 'fuelStatus' + VEHICLE_LIGHTS = 'vehicleLights' + LV_BATTERY = 'lvBattery' + READINESS = 'readiness' + VEHICLE_HEALTH_INSPECTION = 'vehicleHealthInspection' + VEHICLE_HEALTH_WARNINGS = 'vehicleHealthWarnings' + OIL_LEVEL = 'oilLevel' + MEASUREMENTS = 'measurements' + BATTERY_SUPPORT = 'batterySupport' + PARKING = 'parking' + TRIPS = 'trips' diff --git a/wci/wci.py b/wci/wci.py new file mode 100644 index 0000000..c8d8988 --- /dev/null +++ b/wci/wci.py @@ -0,0 +1,89 @@ +import json +import random +import re +import requests +import string + +#BASE_URL = 'https://mobileapi.apps.emea.vwapps.io' +BASE_URL = 'https://emea.bff.cariad.digital/vehicle/v1' +LOGIN_URL= 'https://emea.bff.cariad.digital/user-login/v1' + +def generate_url(): + nonce = ''.join(random.choices(string.ascii_uppercase + string.ascii_lowercase + string.digits, k=16)) + + return LOGIN_URL + '/authorize?nonce=' + nonce + '&redirect_uri=weconnect://authenticated' + +class WeConnectId: + def __init__(self, email_address, password, access_token=None): + self._email_address = email_address + self._password = password + self._access_token = access_token + + self._setup_session() + self._sign_in() + + def _setup_session(self): + self._session = requests.session() + + self._session.headers.update({ + 'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.45 Safari/537.36', + }) + + def _sign_in(self, force=False): + if self._access_token is None or force: + r = self._session.get(generate_url()) + + # Enter e-mail address + post_data = { + '_csrf': re.search('name="_csrf" value="([^"]+)"', r.text).group(1), + 'relayState': re.search('name="relayState" value="([^"]+)"', r.text).group(1), + 'hmac': re.search('name="hmac" value="([^"]+)"', r.text).group(1), + 'email': self._email_address, + } + + uuid = re.search('action="/signin-service/v1/([^@]+)@apps_vw-dilab_com/login/identifier"', r.text).group(1) + + r = self._session.post( + 'https://identity.vwgroup.io/signin-service/v1/' + uuid + '@apps_vw-dilab_com/login/identifier', + data=post_data + ) + + some_data = json.loads(re.search('templateModel: ({.*}),', r.text).group(1)) + + # Enter password + post_data = { + '_csrf': re.search('csrf_token: \'([^\']+)\'', r.text).group(1), + 'relayState': some_data['relayState'], + 'hmac': some_data['hmac'], + 'email': self._email_address, + 'password': self._password, + } + + access_token = None + + try: + r = self._session.post( + 'https://identity.vwgroup.io/signin-service/v1/' + some_data['clientLegalEntityModel']['clientId'] + '/login/authenticate', + data=post_data + ) + + except requests.exceptions.InvalidSchema as e: + # We expect a redirect to 'weconnect://authenticated', which requests doesn't understand + self._access_token = re.search('access_token=([^&$]+)', str(e)).group(1) + + self._session.headers.update({ + 'Authorization': 'Bearer ' + self._access_token, + }) + + def get_access_token(self): + return self._access_token + + def get(self, endpoint): + r = self._session.get(BASE_URL + endpoint) + + if r.status_code == 401: + self._sign_in(True) + + r = self._session.get(BASE_URL + endpoint) + + return r.json()