diff --git a/srv/ohme.py b/ohme/__init__.py similarity index 100% rename from srv/ohme.py rename to ohme/__init__.py diff --git a/ohme/firebase.py b/ohme/firebase.py new file mode 100644 index 0000000..55f54ec --- /dev/null +++ b/ohme/firebase.py @@ -0,0 +1,95 @@ +import os +from typing import Any + +from urllib.parse import urljoin +from dotenv_vault import load_dotenv + +from .make_requests import post_dict + +load_dotenv() + + +def get_firebase_auth_headers() -> dict[str, str]: + """ + Generate authentication headers required for Firebase API requests. + + The headers are constructed based on environment variables which should have been + set prior to calling this function. The required environment variables are: + - ohme_firebase_token + - ohme_firebase_installation_token + - ohme_firebase_device_token + + :return: A dictionary containing the authentication headers. + :rtype: dict[str, str] + """ + token = os.environ["ohme_firebase_token"] + installation_token = os.environ["ohme_firebase_installation_token"] + device_token = os.environ["ohme_firebase_device_token"] + return { + "Accept": "*/*", + "X-firebase-client": "apple-platform/ios apple-sdk/20C52 appstore/true deploy/cocoapods device/iPhone15,2 fire-abt/8.2.0 fire-analytics/8.1.1 fire-auth/8.2.0 fire-dl/8.2.0 fire-fcm/8.2.0 fire-install/8.2.0 fire-ios/8.2.0 fire-rc/8.2.0 firebase-crashlytics/8.2.0 os-version/17.0.3 xcode/14C18", # noqa: E501 + "Authorization": f"AidLogin {device_token}:{installation_token}", + "X-firebase-client-log-type": "0", + "Accept-Encoding": "gzip, deflate, br", + "Accept-Language": "en-GB,en;q=0.9", + "app": "io.ohme.ios.OhmE", + "Conetent-Type": "application/x-www-form-urlencoded", + "User-Agent": "Ohme/543 CFNetwork/1474 Darwin/23.0.0", + "Connection": "keep-alive", + "info": "", + "x-goog-firebase-installations-auth": f"{token}", + } + + +def get_firebase_auth_body() -> dict[str, str]: + """ + Generate the request body required for Firebase API requests. + + The body is constructed based on an environment variable which should have been + set prior to calling this function. The required environment variable is: + - ohme_firebase_device_token + + :return: A dictionary containing the key-value pairs for the request body. + :rtype: dict[str, str] + """ + device_token = os.environ["ohme_firebase_device_token"] + return { + "X-osv": "17.0.3", + "device": device_token, + "X-scope": "*", + "plat": "2", + "app": "io.ohme.ios.OhmE", + "app_ver": "1.28.1", + "X-cliv": "fiid-8.2.0", + "sender": "206163667850", + "X-subtype": "206163667850", + "appid": "f1TW3-vVsEbwuIuDui2MoQ", + "gmp_app_id": "1:206163667850:ios:6f2cd746818dd6de", + } + + +def get_firebase_token() -> str: + """ + Retrieve a Firebase token by sending a HTTP request to the Firebase API. + + This function calls `get_firebase_auth_headers` and `get_firebase_auth_body` to + obtain the necessary headers and body for the request, then sends a HTTP request + to the Firebase API to obtain a Firebase token. + + :return: The Firebase token retrieved from the API. + :rtype: str + :raises Exception: If there is an error retrieving the token from the API. + """ + headers = get_firebase_auth_headers() + body = get_firebase_auth_body() + request_url = "https://fcmtoken.googleapis.com/register" + response = post_dict(request_url, headers, body) + # body looks like token=f1TW3-vVsEbwuIuDui2MoQ:APA91b... + # or error=Missing+registration+token + # 1. Remove token= + # 2. Split on = + # 3. Return the second element + if response.text.split("=")[0] == "token": + return response.text.split("=")[1] + else: + raise Exception(f"Error getting firebase token: {response.text}") diff --git a/ohme/make_requests.py b/ohme/make_requests.py new file mode 100644 index 0000000..21b6257 --- /dev/null +++ b/ohme/make_requests.py @@ -0,0 +1,30 @@ +import httpx +import logging + +logger = logging.getLogger(__name__) + + +class SendRequestError(Exception): + """Exception raised for errors in the send_request function.""" + + pass + + +def post_dict( + url: str, headers: dict[str, str], body: dict[str, str] +) -> httpx.Response: + try: + request = httpx.Request("POST", url=url, headers=headers, data=body) + + with httpx.Client(timeout=10.0) as client: + response = client.send(request) + response.raise_for_status() + return response + except httpx.HTTPError as e: + logger.exception(e) + raise SendRequestError( + f"HTTP error occurred while sending request to {url}" + ) from e + except Exception as e: + logger.exception(e) + raise e diff --git a/test/__init__.py b/test/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/test/test_firebase.py b/test/test_firebase.py new file mode 100644 index 0000000..a02a6ce --- /dev/null +++ b/test/test_firebase.py @@ -0,0 +1,109 @@ +import os +import pytest +from unittest.mock import Mock + +from ohme.firebase import ( + get_firebase_auth_headers, + get_firebase_auth_body, + get_firebase_token, +) + + +class TestGetFirebaseAuthHeaders: + @pytest.fixture(autouse=True) + def setup_method(self, monkeypatch): + # Mock environment variables + monkeypatch.setenv("ohme_firebase_token", "test_token") + monkeypatch.setenv( + "ohme_firebase_installation_token", "test_installation_token" + ) + monkeypatch.setenv("ohme_firebase_device_token", "test_device_token") + + def test_get_firebase_auth_headers(self): + headers = get_firebase_auth_headers() + + assert headers["Accept"] == "*/*" + assert ( + headers["Authorization"] + == "AidLogin test_device_token:test_installation_token" + ) + assert headers["x-goog-firebase-installations-auth"] == "test_token" + # skipping the other headers for brevity + + def test_missing_environment_variable(self, monkeypatch): + monkeypatch.delenv("ohme_firebase_token", raising=False) + + with pytest.raises(KeyError): + get_firebase_auth_headers() + + +class TestGetFirebaseAuthBody: + @pytest.fixture(autouse=True) + def setup_method(self, monkeypatch): + # Mock environment variable + monkeypatch.setenv("ohme_firebase_device_token", "test_device_token") + + def test_device_token_mapping(self): + body = get_firebase_auth_body() + assert body["device"] == "test_device_token" + + def test_missing_environment_variable(self, monkeypatch): + monkeypatch.delenv("ohme_firebase_device_token", raising=False) + + with pytest.raises(KeyError): + get_firebase_auth_body() + + +class TestGetFirebaseToken: + @pytest.fixture(autouse=True) + def setup_method(self, monkeypatch): + # Mock get_firebase_auth_headers and get_firebase_auth_body + self.mock_get_firebase_auth_headers = Mock(return_value={"mocked": "headers"}) + self.mock_get_firebase_auth_body = Mock(return_value={"mocked": "body"}) + monkeypatch.setattr( + "ohme.firebase.get_firebase_auth_headers", + self.mock_get_firebase_auth_headers, + ) + monkeypatch.setattr( + "ohme.firebase.get_firebase_auth_body", self.mock_get_firebase_auth_body + ) + + def test_successful_token_retrieval(self, monkeypatch): + # Mock send_request for a successful response + mock_send_request = Mock() + mock_send_request.return_value.text = "token=sample_token" + monkeypatch.setattr("ohme.firebase.send_request", mock_send_request) + + token = get_firebase_token() + assert token == "sample_token" + + # Check that the mocked functions were called with the correct parameters + self.mock_get_firebase_auth_headers.assert_called() + self.mock_get_firebase_auth_body.assert_called() + mock_send_request.assert_called_with( + "https://fcmtoken.googleapis.com/register", + {"mocked": "headers"}, + {"mocked": "body"}, + ) + + def test_error_response(self, monkeypatch): + # Mock send_request for an error response + mock_error_response = Mock() + mock_error_response.return_value.text = "error=Missing+registration+token" + monkeypatch.setattr("ohme.firebase.send_request", mock_error_response) + + with pytest.raises(Exception) as exc_info: + get_firebase_token() + assert ( + str(exc_info.value) + == "Error getting firebase token: error=Missing+registration+token" + ) + + # Check that the mocked functions were called with the correct parameters + self.mock_get_firebase_auth_headers.assert_called() + self.mock_get_firebase_auth_body.assert_called() + mock_error_response.assert_called_with( + "https://fcmtoken.googleapis.com/register", + {"mocked": "headers"}, + {"mocked": "body"}, + ) diff --git a/test/test_make_requests.py b/test/test_make_requests.py new file mode 100644 index 0000000..58e68a0 --- /dev/null +++ b/test/test_make_requests.py @@ -0,0 +1,75 @@ +import pytest +import httpx +from ohme.make_requests import post_dict, SendRequestError +from unittest.mock import MagicMock, patch + + +class TestPostDict: + @pytest.fixture(autouse=True) + def setup_method(self, monkeypatch): + # Mock the httpx.Request constructor + self.mock_request = MagicMock() + monkeypatch.setattr("httpx.Request", lambda *args, **kwargs: self.mock_request) + + def test_successful_post(self, monkeypatch): + # Mock the httpx.Client.send method to return a successful response + mock_response = MagicMock() + mock_response.raise_for_status.return_value = None # No exception is raised + monkeypatch.setattr("httpx.Client.send", lambda self, request: mock_response) + + response = post_dict( + "https://example.com", {"Header": "Value"}, {"Param": "Value"} + ) + + # Assert the response is the mocked response + assert response == mock_response + + def test_http_error(self, monkeypatch): + # Mock the httpx.Client.send method to raise an httpx.HTTPError + mock_response = MagicMock() + mock_response.raise_for_status.side_effect = httpx.HTTPError("HTTP Error") + monkeypatch.setattr("httpx.Client.send", lambda self, request: mock_response) + + with pytest.raises(SendRequestError) as exc_info: + post_dict("https://example.com", {"Header": "Value"}, {"Param": "Value"}) + assert ( + str(exc_info.value) + == "HTTP error occurred while sending request to https://example.com" + ) + + def test_other_error(self, monkeypatch): + # Mock the httpx.Client.send method to raise a generic exception + monkeypatch.setattr( + "httpx.Client.send", lambda self, request: 1 / 0 + ) # This will raise a ZeroDivisionError + + with pytest.raises(ZeroDivisionError): + post_dict("https://example.com", {"Header": "Value"}, {"Param": "Value"}) + + +class TestPostDict: + @patch.object(httpx, "Request") + @patch.object(httpx.Client, "send") + def test_request_parameters(self, mock_send, mock_request): + # Mock the response + mock_response = MagicMock() + mock_response.raise_for_status.return_value = None + mock_send.return_value = mock_response + + # Mock the request + mock_request_instance = MagicMock() + mock_request.return_value = mock_request_instance + + # Call the function + post_dict("https://example.com", {"Header": "Value"}, {"Param": "Value"}) + + # Check the parameters passed to httpx.Request constructor + mock_request.assert_called_once_with( + "POST", + url="https://example.com", + headers={"Header": "Value"}, + data={"Param": "Value"}, + ) + + # Check that the mocked request object was passed to the send() method + mock_send.assert_called_once_with(mock_request_instance)