Proxying works!

nanomad · nanomad · commit f9f54962c72f · 2024-09-28T16:43:27.000+02:00
diff --git a/main.py b/main.py
@@ -4,7 +4,7 @@
 
 import httpx
 from flask import Flask, request, make_response
-from saic_ismart_client_ng.net.crypto import decrypt_request, encrypt_response
+from saic_ismart_client_ng.net.crypto import decrypt_request, encrypt_response, encrypt_response_full
 from saic_ismart_client_ng.net.httpx import encrypt_httpx_request, decrypt_httpx_response
 
 app = Flask(__name__)
@@ -25,92 +25,122 @@ async def encrypt_httpx_request_wrapper(
     )
 
 
+async def passtrough(path):
+    passtrough_client = httpx.AsyncClient()
+    headers = dict(request.headers)
+    params = request.args
+    if request.method in ['POST', 'PUT']:
+        content = request.get_data(parse_form_data=False).decode('utf-8')
+    else:
+        content = None
+    headers.pop('Host')
+    api_response = await passtrough_client.request(
+        url=f'{base_uri}{path}',
+        method=request.method,
+        params=params,
+        headers=headers,
+        data=content,
+    )
+    api_response_headers = api_response.headers
+    api_response_content = api_response.content
+    api_response_code = api_response.status_code
+    response = make_response(api_response_content)
+    response.headers.update(api_response_headers.items())
+    return response, api_response_code
+
+
 @app.route('/api.app/v1/', defaults={'path': ''})
 @app.route('/api.app/v1/<path:path>', methods=['GET'])
-def do_get(path):
-    print(request)
-    return 'You want GET path: %s' % path
+async def do_get(path):
+    print(f"do_get {path}")
+    return await passtrough(path)
 
 
 @app.route('/api.app/v1/', defaults={'path': ''}, methods=['POST'])
 @app.route('/api.app/v1/<path:path>', methods=['POST'])
 async def do_post(path):
+    print(f"do_post {path}")
     if path == 'oauth/token':
-        raw_data = request.get_data(parse_form_data=False).decode('utf-8')
-        decrypted = decrypt_request(
-            original_request_url=request.url,
-            original_request_headers=request.headers,
-            original_request_content=raw_data,
-            base_uri=request.url.removesuffix(path),
+        return await handle_login(path)
+    else:
+        return await passtrough(path)
+
+
+async def handle_login(path):
+    decrypting_client = httpx.AsyncClient(
+        event_hooks={
+            "request": [encrypt_httpx_request_wrapper],
+            "response": [decrypt_httpx_response]
+        },
+    )
+    raw_data = request.get_data(parse_form_data=False).decode('utf-8')
+    decrypted = decrypt_request(
+        original_request_url=request.url,
+        original_request_headers=request.headers,
+        original_request_content=raw_data,
+        base_uri=request.url.removesuffix(path),
+    )
+    unquoted = urllib.parse.parse_qs(decrypted)
+    username = unquoted[b'username'][0].decode('utf-8')
+    password = unquoted[b'password'][0].decode('utf-8')
+    device_id = unquoted[b'deviceId'][0].decode('utf-8')
+    device_type = unquoted[b'deviceType'][0].decode('utf-8')
+    scope = unquoted[b'scope'][0].decode('utf-8')
+    grant_type = unquoted[b'grant_type'][0].decode('utf-8')
+    login_type = unquoted[b'loginType'][0].decode('utf-8')
+    country_code = unquoted[b'countryCode'][0].decode('utf-8') if b'countryCode' in unquoted else ''
+    headers = {
+        "Content-Type": "application/x-www-form-urlencoded",
+        "Accept": "application/json",
+        "Authorization": request.headers['Authorization']
+    }
+    form_body = {
+        "grant_type": grant_type,
+        "username": username,
+        "password": password,
+        "scope": scope,
+        "deviceId": device_id,
+        "deviceType": device_type,
+        "loginType": login_type,
+        "countryCode": country_code
+    }
+    api_response = await decrypting_client.post(url=f'{base_uri}{path}', data=form_body, headers=headers)
+    if api_response.is_success:
+        cached_tokens[username] = api_response.text
+        ts = api_response.headers['app-send-date']
+        new_content, new_headers = encrypt_response_full(
+            original_request_url=str(api_response.url),
+            original_response_headers=api_response.headers,
+            original_response_content=api_response.text,
+            response_timestamp_ms=ts,
+            base_uri=base_uri,
+            tenant_id='459771',
+            user_token=''
         )
-        unquoted = urllib.parse.parse_qs(decrypted)
-        username = unquoted[b'username'][0].decode('utf-8')
-        if username in cached_tokens:
-            response = cached_tokens[username]
-        else:
-            password = unquoted[b'password'][0].decode('utf-8')
-            login_type = unquoted[b'loginType'][0].decode('utf-8')
-            country_code = unquoted[b'countryCode'][0].decode('utf-8') if b'countryCode' in unquoted else ''
-            headers = {
-                "Content-Type": "application/x-www-form-urlencoded",
-                "Accept": "application/json",
-                "Authorization": request.headers['Authorization']
-            }
-            firebase_device_id = "cqSHOMG1SmK4k-fzAeK6hr:APA91bGtGihOG5SEQ9hPx3Dtr9o9mQguNiKZrQzboa-1C_UBlRZYdFcMmdfLvh9Q_xA8A0dGFIjkMhZbdIXOYnKfHCeWafAfLXOrxBS3N18T4Slr-x9qpV6FHLMhE9s7I6s89k9lU7DD"
-            form_body = {
-                "grant_type": "password",
-                "username": username,
-                "password": password,
-                "scope": "all",
-                "deviceId": f"{firebase_device_id}###europecar",
-                "deviceType": "1",  # 2 for huawei
-                "loginType": login_type,
-                "countryCode": country_code
-            }
-            client = httpx.AsyncClient(
-                event_hooks={
-                    "request": [encrypt_httpx_request_wrapper],
-                    "response": [decrypt_httpx_response]
-                },
-            )
-            response = await client.post(url=f'{base_uri}{path}', data=form_body, headers=headers)
-            if response.is_success:
-                response_json = response.json()
-                cached_tokens[username] = response_json
-                text_content = json.dumps(response_json)
-                ts = response.headers['app-send-date']
-                their_verification = response.headers['app-verification-string']
-                new_content, new_headers = encrypt_response(
-                    original_request_url=str(response.url),
-                    original_response_headers=response.headers,
-                    original_response_content=response.text,
-                    response_timestamp_ms=ts,
-                    base_uri=base_uri,
-                    tenant_id='459771',
-                    user_token=''
-                )
-                response = make_response(new_content)
-                response.headers.update(new_headers)
-                return response
-            else:
-                return (response.status_code, response.content)
-
-    return 'You want POST path: %s' % path
+        response = make_response(new_content)
+        generated_headers = dict(new_headers.items())
+        generated_headers.pop('content-length')
+        response.headers.update(generated_headers)
+        return response, api_response.status_code
+    else:
+        response = make_response(api_response.text)
+        response.headers.update(api_response.headers.items())
+        return response, api_response.status_code
 
 
 @app.route('/api.app/v1/', defaults={'path': ''}, methods=['DELETE'])
 @app.route('/api.app/v1/<path:path>', methods=['DELETE'])
-def do_delete(path):
-    print(request)
-    return 'You want GET path: %s' % path
+async def do_delete(path):
+    print(f"do_delete {path}")
+    return await passtrough(path)
 
 
 @app.route('/api.app/v1/', defaults={'path': ''}, methods=['PUT'])
 @app.route('/api.app/v1/<path:path>', methods=['PUT'])
-def do_put(path):
-    print(request)
-    return 'You want POST path: %s' % path
+async def do_put(path):
+    print(f"do_put {path}")
+    return await passtrough(path)
 
 
 if __name__ == '__main__':
-    app.run()
+    app.run(host="127.0.0.1", port=8080, debug=True)
diff --git a/pyproject.toml b/pyproject.toml
@@ -0,0 +1,78 @@
+[tool.poetry]
+name = "saic_python_api_proxy"
+homepage = "https://github.com/SAIC-iSmart-API/saic-python-api-proxy"
+version = "0.2.3"
+description = "MG iSMART API proxy"
+authors = [
+    "Giovanni Condello <saic-python-client@nanomad.net>",
+]
+readme = "README.md"
+package-mode = false
+classifiers = [
+    "Programming Language :: Python :: 3",
+    "License :: OSI Approved :: MIT License",
+    "Operating System :: OS Independent",
+]
+
+[tool.poetry.urls]
+"Bug Tracker" = "https://github.com/SAIC-iSmart-API/saic-python-api-proxy/issues"
+
+
+[tool.poetry.dependencies]
+python = "^3.11"
+httpx = "^0.27.0"
+flask = { version = "^3.0.3", extras = ["async"] }
+saic_ismart_client_ng = { git = "https://github.com/SAIC-iSmart-API/saic-python-client-ng" }
+
+[tool.poetry.dev-dependencies]
+pytest = "^8.2.2"
+mock = "^5.1.0"
+coverage = "^7.5.4"
+ruff = "^0.4.10"
+pytest-cov = "^5.0.0"
+pytest-asyncio = "^0.23.7"
+pytest-mock = "^3.14.0"
+
+
+
+[build-system]
+requires = ["poetry-core>=1.0.0"]
+build-backend = "poetry.core.masonry.api"
+
+[tool.pytest.ini_options]
+norecursedirs = ".git build dist"
+testpaths = "tests"
+#pythonpath = [
+#    "src"
+#]
+mock_use_standalone_module = true
+addopts = [
+    "--import-mode=importlib",
+]
+
+[tool.coverage.run]
+omit = [
+    "tests/*",
+]
+branch = true
+command_line = "-m pytest"
+
+[tool.coverage.report]
+# Regexes for lines to exclude from consideration
+exclude_lines = [
+    # Have to re-enable the standard pragma
+    'pragma: no cover',
+    # Don't complain about missing debug-only code:
+    'def __repr__',
+    'if self\.debug',
+    # Don't complain if tests don't hit defensive assertion code:
+    'raise AssertionError',
+    'raise NotImplementedError',
+    # Don't complain if non-runnable code isn't run:
+    'if 0:',
+    'if __name__ == .__main__.:',
+]
+ignore_errors = true
+
+[tool.ruff]
+output-format = "github"
