clean up code slightly

Author: Ian-Nara

tests/test_publisher_client.py

@@ -0,0 +1,192 @@
+import os
+import unittest
+
+from uid2_client import Uid2PublisherClient
+from uid2_client import TokenGenerateInput
+from uid2_client import TokenGenerateResponse
+from uid2_client.identity_tokens import IdentityTokens
+from urllib.request import HTTPError
+
+
+class PublisherEuidIntegrationTests(unittest.TestCase):
+
+    EUID_SECRET_KEY = None
+    EUID_API_KEY = None
+    EUID_BASE_URL = None
+
+    publisher_client = None
+
+    @classmethod
+    def setUpClass(cls):
+        cls.EUID_BASE_URL = os.getenv("EUID_BASE_URL")
+        cls.EUID_API_KEY = os.getenv("EUID_API_KEY")
+        cls.EUID_SECRET_KEY = os.getenv("EUID_SECRET_KEY")
+
+        print(cls.EUID_BASE_URL, cls.EUID_API_KEY, cls.EUID_SECRET_KEY)
+
+        if cls.EUID_BASE_URL and cls.EUID_API_KEY and cls.EUID_SECRET_KEY:
+            cls.publisher_client = Uid2PublisherClient(cls.EUID_BASE_URL, cls.EUID_API_KEY, cls.EUID_SECRET_KEY)
+
+    def test_integration_tc_string(self):
+        tc_string = "CPhJRpMPhJRpMABAMBFRACBoALAAAEJAAIYgAKwAQAKgArABAAqAAA"
+
+        token_generate_response = self.publisher_client.generate_token(
+            TokenGenerateInput.from_email("[email protected]").with_transparency_and_consent_string(tc_string))
+        self.assertFalse(token_generate_response.is_optout())
+
+        identity = token_generate_response.get_identity()
+        self.assertIsNotNone(identity)
+        self.assertFalse(identity.is_due_for_refresh())
+        self.assertIsNotNone(identity.get_advertising_token())
+        self.assertIsNotNone(identity.get_refresh_token())
+        self.assertIsNotNone(identity.get_json_string())
+        self.assertTrue(identity.is_refreshable())
+
+    def test_integration_tc_string_with_insufficient_consent(self):
+        tc_string = "CPehXK9PehXK9ABAMBFRACBoADAAAEJAAIYgAKwAQAKgArABAAqAAA"
+        with self.assertRaises(ValueError):
+            self.publisher_client.generate_token(TokenGenerateInput.from_email("[email protected]").with_transparency_and_consent_string(tc_string))
+
+
+    def test_integration_optout_generate_token(self):
+        publisher_client = Uid2PublisherClient(self.EUID_BASE_URL, self.EUID_API_KEY, self.EUID_SECRET_KEY)
+        tc_string = "CPhJRpMPhJRpMABAMBFRACBoALAAAEJAAIYgAKwAQAKgArABAAqAAA"
+        input = TokenGenerateInput.from_email("[email protected]").do_not_generate_tokens_for_opted_out().with_transparency_and_consent_string(tc_string)
+        token_generate_response = publisher_client.generate_token(input)
+        self.assertTrue(token_generate_response.is_optout())
+        self.assertFalse(token_generate_response.is_success())
+        self.assertIsNone(token_generate_response.get_identity())
+
+class PublisherUid2IntegrationTests(unittest.TestCase):
+
+    UID2_SECRET_KEY = None
+    UID2_API_KEY = None
+    UID2_BASE_URL = None
+
+    publisher_client = None
+
+    @classmethod
+    def setUpClass(cls):
+        cls.UID2_BASE_URL = os.getenv("UID2_BASE_URL")
+        cls.UID2_API_KEY = os.getenv("UID2_API_KEY")
+        cls.UID2_SECRET_KEY = os.getenv("UID2_SECRET_KEY")
+
+        print(cls.UID2_BASE_URL, cls.UID2_API_KEY, cls.UID2_SECRET_KEY)
+
+        if cls.UID2_BASE_URL and cls.UID2_API_KEY and cls.UID2_SECRET_KEY:
+            cls.publisher_client = Uid2PublisherClient(cls.UID2_BASE_URL, cls.UID2_API_KEY, cls.UID2_SECRET_KEY)
+
+    # Test methods
+    def test_integration_generate_and_refresh(self):
+
+        token_generate_response = self.publisher_client.generate_token(
+            TokenGenerateInput.from_email("[email protected]"))
+
+        self.assertFalse(token_generate_response.is_optout())
+
+        identity = token_generate_response.get_identity()
+        self.assertIsNotNone(identity)
+        self.assertFalse(identity.is_due_for_refresh())
+        self.assertIsNotNone(identity.get_advertising_token())
+        self.assertIsNotNone(identity.get_refresh_token())
+        self.assertIsNotNone(identity.get_json_string())
+        self.assertTrue(identity.is_refreshable())
+
+        token_refresh_response = self.publisher_client.refresh_token(identity)
+        self.assertTrue(token_refresh_response.is_success())
+        self.assertFalse(token_refresh_response.is_optout())
+        self.assertIsNotNone(token_refresh_response.get_identity_json_string())
+
+        refreshed_identity = token_refresh_response.get_identity()
+        self.assertIsNotNone(refreshed_identity)
+        self.assertFalse(refreshed_identity.is_due_for_refresh())
+        self.assertIsNotNone(refreshed_identity.get_advertising_token())
+        self.assertIsNotNone(refreshed_identity.get_refresh_token())
+        self.assertIsNotNone(refreshed_identity.get_json_string())
+        self.assertTrue(identity.is_refreshable())
+
+    def test_integration_optout(self):
+
+        token_generate_response = self.publisher_client.generate_token(TokenGenerateInput.from_email("[email protected]"))
+
+        self.assertFalse(token_generate_response.is_optout())
+
+        identity = token_generate_response.get_identity()
+        self.assertIsNotNone(identity)
+        self.assertFalse(identity.is_due_for_refresh())
+        self.assertIsNotNone(identity.get_advertising_token())
+        self.assertIsNotNone(identity.get_refresh_token())
+        self.assertIsNotNone(identity.get_json_string())
+        self.assertTrue(identity.is_refreshable())
+
+        token_refresh_response = self.publisher_client.refresh_token(identity)
+        self.assertFalse(token_refresh_response.is_success())
+        self.assertTrue(token_refresh_response.is_optout())
+        self.assertIsNone(token_refresh_response.get_identity_json_string())
+        self.assertIsNone(token_refresh_response.get_identity())
+
+    def test_integration_phone(self):
+
+        token_generate_response = self.publisher_client.generate_token(
+            TokenGenerateInput.from_phone("+12345678901"))
+
+        self.assertFalse(token_generate_response.is_optout())
+        identity = token_generate_response.get_identity()
+        self.assertIsNotNone(identity)
+        self.assertFalse(identity.is_due_for_refresh())
+        self.assertIsNotNone(identity.get_advertising_token())
+        self.assertIsNotNone(identity.get_refresh_token())
+        self.assertIsNotNone(identity.get_json_string())
+        self.assertTrue(identity.is_refreshable())
+
+        token_refresh_response = self.publisher_client.refresh_token(identity)
+        self.assertTrue(token_refresh_response.is_success())
+        self.assertFalse(token_refresh_response.is_optout())
+        self.assertIsNotNone(token_refresh_response.get_identity_json_string())
+
+        refreshed_identity = token_refresh_response.get_identity()
+        self.assertIsNotNone(refreshed_identity)
+        self.assertFalse(refreshed_identity.is_due_for_refresh())
+        self.assertIsNotNone(refreshed_identity.get_advertising_token())
+        self.assertIsNotNone(refreshed_identity.get_refresh_token())
+        self.assertIsNotNone(refreshed_identity.get_json_string())
+        self.assertTrue(identity.is_refreshable())
+
+    def test_integration_bad_requests(self):
+
+        with self.assertRaises(ValueError):
+            self.publisher_client.generate_token(TokenGenerateInput.from_email("this is not an email address"))
+
+        with self.assertRaises(ValueError):
+            self.publisher_client.generate_token(TokenGenerateInput.from_phone("this is not a phone number"))
+
+        unnormalized_phone_number = " +123 44 55-66-77"
+        with self.assertRaises(ValueError):
+            self.publisher_client.generate_token(TokenGenerateInput.from_phone(unnormalized_phone_number))
+
+        expired_respose = "{\"advertising_token\":\"AgAAAAN6QZRCFTau+sfOlMMUY2ftElFMq2TCrcu1EAaD9WmEfoT2BWm2ZKz1tumbT00tWLffRDQ/9POXfA0O/Ljszn7FLtG5EzTBM3HYs4f5irkqeEvu38DhVCxUEpI+gZZZkynRap1oYx6AmC/ip3rk+7pmqa3r3saDs1mPRSSTm+Nh6A==\",\"user_token\":\"AgAAAAL6aleYI4BubI5ZXMBshqmMEfCkbCJF4fLeg1sdI0BTLzj9sXsSISjkG0lMC743diC2NVy3ElkbO1lLysd+Lm6alkqevPrcuWDisQ1939YdoH6LqpwBH3FNSE4/xa3Q+94=\",\"refresh_token\":\"AAAAAARomrP3NjjH+8mt5djfTHbmRZXjOMnAN8WpjJoe30AhUCvYksO/xoDSj77GzWv4M99DhnPl2cVco8CZFTcE10nauXI4Barr890ILnH0IIacOei5Zjwh6DycFkoXkAAuHY1zjmxb7niGLfSP2RctWkZdRVGWQv/UW/grw6+paU9bnKEWPzVvLwwdW2NgjDKu+szE6A+b5hkY+I3voKoaz8/kLDmX8ddJGLy/YOh/LIveBspSAvEg+v89OuUCwAqm8L3Rt8PxDzDnt0U4Na+AUawvvfsIhmsn/zMpRRks6GHhIAB/EQUHID8TedU8Hv1WFRsiraG9Dfn1Kc5/uYnDJhEagWc+7RgTGT+U5GqI6+afrAl5091eBLbmvXnXn9ts\",\"identity_expires\":1668059799628,\"refresh_expires\":1668142599628,\"refresh_from\":1668056202628,\"refresh_response_key\":\"P941vVeuyjaDRVnFQ8DPd0AZnW4bPeiJPXER2K9QXcU=\"}"
+        current_identity = IdentityTokens.from_json_string(expired_respose)
+        with self.assertRaises(HTTPError):
+            self.publisher_client.refresh_token(current_identity)
+
+        with self.assertRaises(TypeError):
+            self.publisher_client.generate_token(TokenGenerateInput.from_email(None))
+
+        with self.assertRaises(AttributeError):
+            self.publisher_client.refresh_token(None)
+
+        bad_url_client = Uid2PublisherClient("https://www.something.com", self.UID2_API_KEY, self.UID2_SECRET_KEY)
+        with self.assertRaises(HTTPError):
+            bad_url_client.generate_token(TokenGenerateInput.from_email("[email protected]"))
+
+        bad_secret_client = Uid2PublisherClient(self.UID2_BASE_URL, self.UID2_API_KEY, "badSecretKeypB64Y3fV2dAed8t/mupw3sjN5jNRFzg=")
+        with self.assertRaises(HTTPError):
+            bad_secret_client.generate_token(TokenGenerateInput.from_email("[email protected]"))
+
+        bad_api_client = Uid2PublisherClient(self.UID2_BASE_URL, "not-real-key", self.UID2_SECRET_KEY)
+        with self.assertRaises(HTTPError):
+            bad_secret_client.generate_token(TokenGenerateInput.from_email("[email protected]"))
+
+
+if __name__ == '__main__':
+    unittest.main()

uid2_client/__init__.py

@@ -14,5 +14,8 @@
 from .keys import *
 from .euid_client_factory import *
 from .uid2_client_factory import *
+from .token_generate_input import *
+from .token_generate_response import *
+from .publisher_client import *
 
 

uid2_client/identity_tokens.py

@@ -0,0 +1,78 @@
+import datetime
+import json
+
+
+class IdentityTokens:
+    def __init__(self, advertising_token, refresh_token, refresh_response_key, identity_expires,
+                 refresh_expires, refresh_from, json_string):
+        self._advertising_token = advertising_token
+        self._refresh_token = refresh_token
+        self._refresh_response_key = refresh_response_key
+        self._identity_expires = identity_expires
+        self._refresh_expires = refresh_expires
+        self._refresh_from = refresh_from
+        self._json_string = json_string
+
+    @staticmethod
+    def from_json_string(json_string):
+        assert json_string is not None, "jsonString must not be null"
+
+        try:
+            return IdentityTokens.from_json(json.loads(json_string))
+        except json.JSONDecodeError:
+            print("Invalid json string")
+            return None
+        except KeyError:
+            print("Missing field in json string")
+            return None
+
+    @staticmethod
+    def from_json(json_obj):
+        return IdentityTokens(
+            json_obj.get("advertising_token"),
+            json_obj.get("refresh_token"),
+            json_obj.get("refresh_response_key"),
+            json_obj.get("identity_expires"),
+            json_obj.get("refresh_expires"),
+            json_obj.get("refresh_from"),
+            json.dumps(json_obj)
+        )
+
+    def is_due_for_refresh(self):
+        return self.is_due_for_refresh_impl(datetime.datetime.now())
+
+    def get_advertising_token(self):
+        return self._advertising_token
+
+    def get_refresh_token(self):
+        return self._refresh_token
+
+    def get_json_string(self):
+        return self._json_string
+
+    def is_refreshable(self):
+        return self.is_refreshable_impl(datetime.datetime.now())
+
+    def is_refreshable_impl(self, timestamp):
+        refresh_expires = self._refresh_expires
+        if refresh_expires is None or timestamp.timestamp() > refresh_expires:
+            return False
+        return self._refresh_token is not None
+
+    def is_due_for_refresh_impl(self, timestamp):
+        return timestamp.timestamp() > self._refresh_from or self.has_identity_expired(timestamp)
+
+    def has_identity_expired(self, timestamp):
+        return timestamp.timestamp() > self._identity_expires
+
+    def get_refresh_response_key(self):
+        return self._refresh_response_key
+
+    def get_identity_expires(self):
+        return self._identity_expires
+
+    def get_refresh_expires(self):
+        return self._refresh_expires
+
+    def get_refresh_from(self):
+        return self._refresh_from

uid2_client/input_util.py

@@ -0,0 +1,104 @@
+import hashlib
+import base64
+
+
+def is_phone_number_normalized(phone_number):
+    min_phone_number_digits = 10
+    max_phone_number_digits = 15
+
+    if phone_number is None or len(phone_number) < min_phone_number_digits:
+        return False
+
+    if phone_number[0] != '+':
+        return False
+
+    total_digits = sum(char.isdigit() for char in phone_number[1:])
+
+    return min_phone_number_digits <= total_digits <= max_phone_number_digits
+
+
+def normalize_email_string(email):
+    pre_sb = []
+    pre_sb_specialized = []
+    sb = []
+    ws_buffer = []
+
+    class EmailParsingState:
+        Starting = 1
+        Pre = 2
+        SubDomain = 3
+
+    parsing_state = EmailParsingState.Starting
+
+    in_extension = False
+
+    for i in range(len(email)):
+        c_given = email[i]
+        if 'A' <= c_given <= 'Z':
+            c = chr(ord(c_given) + 32)
+        else:
+            c = c_given
+
+        if parsing_state == EmailParsingState.Starting:
+            if c == ' ':
+                continue
+            else:
+                parsing_state = EmailParsingState.Pre
+
+        if parsing_state == EmailParsingState.Pre:
+            if c == '@':
+                parsing_state = EmailParsingState.SubDomain
+            elif c == '.':
+                pre_sb.append(c)
+            elif c == '+':
+                pre_sb.append(c)
+                in_extension = True
+            else:
+                pre_sb.append(c)
+                if not in_extension:
+                    pre_sb_specialized.append(c)
+        elif parsing_state == EmailParsingState.SubDomain:
+            if c == '@':
+                return None
+            elif c == ' ':
+                ws_buffer.append(c)
+                continue
+            if len(ws_buffer) > 0:
+                sb.extend(ws_buffer)
+                ws_buffer = []
+            sb.append(c)
+
+    if len(sb) == 0:
+        return None
+    domain_part = ''.join(sb)
+
+    gmail_domain = "gmail.com"
+    if gmail_domain == domain_part:
+        address_part_to_use = pre_sb_specialized
+    else:
+        address_part_to_use = pre_sb
+
+    if len(address_part_to_use) == 0:
+        return None
+
+    return ''.join(address_part_to_use + ['@'] + [domain_part])
+
+
+def is_ascii_digit(d):
+    return '0' <= d <= '9'
+
+
+def base64_to_byte_array(str):
+    return base64.b64decode(str)
+
+
+def byte_array_to_base64(b):
+    return base64.b64encode(b).decode()
+
+
+def get_base64_encoded_hash(input):
+    return byte_array_to_base64(get_sha256_bytes(input))
+
+
+def get_sha256_bytes(input):
+    return hashlib.sha256(input.encode()).digest()