Merge pull request #52 from IABTechLab/mkc-UID2-3678-base64url

Support base64-encoded v4 tokens

Author: mcollins-ttd

.gitignore

@@ -1,5 +1,11 @@
 *.swp
 .idea/*
 *__pycache__*
-dist/*
-*egg-info/*
+
+# Distribution / packaging
+build/
+dist/
+*egg-info/
+
+# Environments
+.venv

tests/test_bidstream_client.py

@@ -280,5 +280,19 @@ def test_refresh_keys(self, mock_refresh_bidstream_keys):
                                                             client_secret_bytes)
 
 
+    def test_decrypt_v4_token_encoded_as_base64(self):
+        for scope in IdentityScope:
+            with self.subTest(scope=scope):
+                self.refresh(key_bidstream_response_json_default_keys(identity_scope=scope))
+
+                while True:
+                    token = generate_uid_token(scope, AdvertisingTokenVersion.ADVERTISING_TOKEN_V4)
+                    token = base64.b64encode(Uid2Base64UrlCoder.decode(token)).decode('utf-8')
+                    if ("=" in token) and ("/" in token) and ("+" in token):
+                        break
+            
+                self._decrypt_and_assert_success(token, AdvertisingTokenVersion.ADVERTISING_TOKEN_V4, scope)
+
+
 if __name__ == '__main__':
     unittest.main()

tests/test_identity_map_client.py

@@ -7,6 +7,12 @@
 from uid2_client import IdentityMapClient, IdentityMapInput, normalize_and_hash_email, normalize_and_hash_phone
 
 
+@unittest.skipIf(
+    os.getenv("UID2_BASE_URL") == None
+    or os.getenv("UID2_API_KEY") == None
+    or os.getenv("UID2_SECRET_KEY") == None,
+    reason="Environment variables UID2_BASE_URL, UID2_API_KEY, and UID2_SECRET_KEY must be set",
+)
 class IdentityMapIntegrationTests(unittest.TestCase):
     UID2_BASE_URL = None
     UID2_API_KEY = None

tests/test_publisher_client.py

@@ -8,6 +8,12 @@
 from urllib.request import HTTPError
 
 
+@unittest.skipIf(
+    os.getenv("EUID_BASE_URL") == None
+    or os.getenv("EUID_API_KEY") == None
+    or os.getenv("EUID_SECRET_KEY") == None,
+    reason="Environment variables EUID_BASE_URL, EUID_API_KEY, and EUID_SECRET_KEY must be set",
+)
 class PublisherEuidIntegrationTests(unittest.TestCase):
 
     EUID_SECRET_KEY = None
@@ -61,6 +67,12 @@ def test_integration_optout_generate_token(self):
         self.assertFalse(token_generate_response.is_success())
         self.assertIsNone(token_generate_response.get_identity())
 
+@unittest.skipIf(
+    os.getenv("UID2_BASE_URL") == None
+    or os.getenv("UID2_API_KEY") == None
+    or os.getenv("UID2_SECRET_KEY") == None,
+    reason="Environment variables UID2_BASE_URL, UID2_API_KEY, and UID2_SECRET_KEY must be set",
+)
 class PublisherUid2IntegrationTests(unittest.TestCase):
 
     UID2_SECRET_KEY = None

tests/test_sharing_client.py

@@ -3,8 +3,8 @@
 from unittest.mock import patch
 
 from test_utils import *
-from tests.test_bidstream_client import TestBidStreamClient
-from tests.test_encryption import TestEncryptionFunctions
+from test_bidstream_client import TestBidStreamClient
+from test_encryption import TestEncryptionFunctions
 from uid2_client import SharingClient, DecryptionStatus, Uid2ClientFactory
 from uid2_client.encryption_status import EncryptionStatus
 from uid2_client.refresh_response import RefreshResponse
@@ -283,7 +283,7 @@ def test_expiry_in_token_matches_expiry_in_response(self):  # ExpiryInTokenMatch
             key_sharing_response_json([master_key, site_key], identity_scope=IdentityScope.UID2,
                                       default_keyset_id=99999, token_expiry_seconds=2))
 
-        encryption_data_response = self._client.encrypt_raw_uid_into_token(example_uid)
+        encryption_data_response = self._client._encrypt_raw_uid_into_token(example_uid, now=now)
         self.assertEqual(encryption_data_response.status, EncryptionStatus.SUCCESS)
 
         result = self._client._decrypt_token_into_raw_uid(encryption_data_response.encrypted_data, now + dt.timedelta(seconds=1))

tests/test_utils.py

@@ -39,19 +39,18 @@
 phone_uid = "BEOGxroPLdcY7LrSiwjY52+X05V0ryELpJmoWAyXiwbZ"
 
 test_cases_all_scopes_all_versions = [
-    [IdentityScope.UID2, AdvertisingTokenVersion.ADVERTISING_TOKEN_V2],
-    [IdentityScope.UID2, AdvertisingTokenVersion.ADVERTISING_TOKEN_V3],
-    [IdentityScope.UID2, AdvertisingTokenVersion.ADVERTISING_TOKEN_V4],
-    [IdentityScope.EUID, AdvertisingTokenVersion.ADVERTISING_TOKEN_V2],
-    [IdentityScope.EUID, AdvertisingTokenVersion.ADVERTISING_TOKEN_V3],
-    [IdentityScope.EUID, AdvertisingTokenVersion.ADVERTISING_TOKEN_V4]
+    [scope, version]
+    for scope in IdentityScope
+    for version in AdvertisingTokenVersion
 ]
 
 test_cases_all_scopes_v3_v4_versions = [
-    [IdentityScope.UID2, AdvertisingTokenVersion.ADVERTISING_TOKEN_V3],
-    [IdentityScope.UID2, AdvertisingTokenVersion.ADVERTISING_TOKEN_V4],
-    [IdentityScope.EUID, AdvertisingTokenVersion.ADVERTISING_TOKEN_V3],
-    [IdentityScope.EUID, AdvertisingTokenVersion.ADVERTISING_TOKEN_V4]
+    [scope, version]
+    for scope in IdentityScope
+    for version in [
+        AdvertisingTokenVersion.ADVERTISING_TOKEN_V3,
+        AdvertisingTokenVersion.ADVERTISING_TOKEN_V4,
+    ]
 ]
 
 YESTERDAY = now + dt.timedelta(days=-1)

uid2_client/encryption.py

@@ -105,12 +105,25 @@ def _decrypt_token(token, keys, domain_name, client_type, now):
     elif token_bytes[1] == AdvertisingTokenVersion.ADVERTISING_TOKEN_V3.value:
         return _decrypt_token_v3(base64.b64decode(token), keys, domain_name, client_type, now, AdvertisingTokenVersion.ADVERTISING_TOKEN_V3)
     elif token_bytes[1] == AdvertisingTokenVersion.ADVERTISING_TOKEN_V4.value:
-        # same as V3 but use Base64URL encoding
-        return _decrypt_token_v3(Uid2Base64UrlCoder.decode(token), keys, domain_name, client_type, now, AdvertisingTokenVersion.ADVERTISING_TOKEN_V4)
+        # Accept either base64 or base64url encoding.
+        return _decrypt_token_v3(base64.b64decode(_base64url_to_base64(token)), keys, domain_name, client_type, now, AdvertisingTokenVersion.ADVERTISING_TOKEN_V4)
     else:
         return DecryptedToken.make_error(DecryptionStatus.VERSION_NOT_SUPPORTED)
 
 
+def _base64url_to_base64(value):
+    value = value.replace('-', '+').replace('_', '/')
+    input_size_mod4 = len(value) % 4
+    if input_size_mod4 == 0:
+        return value
+    elif input_size_mod4 == 2:
+        return value + '=='
+    elif input_size_mod4 == 3:
+        return value + '='
+    else:
+        raise EncryptionError('invalid payload')
+
+
 def _token_has_valid_lifetime(keys, client_type, generated_or_now, expires, now):
     #  generated_or_now allows "now" for token v2, since v2 does not contain a "token generated" field.
     #  v2 therefore checks against remaining lifetime rather than total lifetime
@@ -294,7 +307,6 @@ def encrypt(uid2, identity_scope, keys, keyset_id=None, **kwargs):
         return EncryptionDataResponse.make_error(EncryptionStatus.ENCRYPTION_FAILURE)
 
 
-
 # DEPRECATED, DO NOT CALL
 def encrypt_data(data, identity_scope, **kwargs):
     """Encrypt arbitrary binary data.

uid2_client/uid2_base64_url_coder.py

@@ -11,18 +11,11 @@ def encode(input):
         encoded_token = base64.urlsafe_b64encode(input).decode('ascii')
         # urlsafe_b64encode doesn't remove the '=' padding per the spec so we should remove it
         # as '=' is a reserved char in URL spec
-        count = 0
-        for i in range(3):
-            if encoded_token[len(encoded_token) - 1 - i] == '=':
-                count = count + 1
-        # encoded_token[:-0] will empty the whole string!
-        if count > 0:
-            return encoded_token[:-count]
-        return encoded_token
+        return encoded_token.rstrip('=')
 
     @staticmethod
     def decode(token):
-        input_size_mod4 = len(token) % 4;
+        input_size_mod4 = len(token) % 4
         if input_size_mod4 > 0:
             padding_needed = 4 - input_size_mod4
             padding = ""