Expose expiry time in decryption response

Author: asloobq

tests/test_bidstream_client.py

@@ -149,6 +149,7 @@ def test_token_expiry_custom_decryption_time(self, mock_refresh_bidstream_keys):
         result = self._client._decrypt_token_into_raw_uid(token, None, expires_at + dt.timedelta(seconds=1))
         self.assertFalse(result.success)
         self.assertEqual(result.status, DecryptionStatus.TOKEN_EXPIRED)
+        self.assertEqual(result.expiry, expires_at)
 
         result = self._client._decrypt_token_into_raw_uid(token, None, expires_at - dt.timedelta(seconds=1))
         self.assertIsNotNone(result)

tests/test_sharing_client.py

@@ -196,6 +196,7 @@ def test_expiry_in_token_matches_expiry_in_response(self, mock_refresh_keys_util
         result = self._client._decrypt_token_into_raw_uid(encryption_data_response.encrypted_data, now + dt.timedelta(seconds=3))
         self.assertFalse(result.success)
         self.assertEqual(DecryptionStatus.TOKEN_EXPIRED, result.status)
+        self.assertEqual(now + dt.timedelta(seconds=2), result.expiry)
 
     def test_encrypt_key_inactive(self, mock_refresh_keys_util):  #EncryptKeyInactive
         def get_post_refresh_keys_response_with_key_inactive():

uid2_client/encryption.py

@@ -157,12 +157,14 @@ def _decrypt_token_v2(token_bytes, keys, domain_name, client_type, now):
     expires_ms = int.from_bytes(master_payload[:8], 'big')
     expires = dt.datetime.fromtimestamp(expires_ms / 1000.0, tz=timezone.utc)
     if expires < now:
-        return DecryptedToken.make_error(DecryptionStatus.TOKEN_EXPIRED)
+        return DecryptedToken(DecryptionStatus.TOKEN_EXPIRED, None, None, None, None,
+                          keys.get_identity_scope(), AdvertisingTokenVersion.ADVERTISING_TOKEN_V2, expires)
 
     site_key_id = int.from_bytes(master_payload[8:12], 'big')
     site_key = keys.get(site_key_id)
     if site_key is None:
-        return DecryptedToken.make_error(DecryptionStatus.NOT_AUTHORIZED_FOR_KEY)
+        return DecryptedToken(DecryptionStatus.NOT_AUTHORIZED_FOR_KEY, None, None, None, site_key.site_id,
+                          keys.get_identity_scope(), AdvertisingTokenVersion.ADVERTISING_TOKEN_V2, expires)
 
     identity_iv = master_payload[12:28]
     identity = _decrypt(master_payload[28:], identity_iv, site_key)
@@ -177,10 +179,11 @@ def _decrypt_token_v2(token_bytes, keys, domain_name, client_type, now):
     established = dt.datetime.fromtimestamp(established_ms / 1000.0, tz=timezone.utc)
 
     if not _token_has_valid_lifetime(keys, client_type, established, expires, now):
-        return DecryptedToken.make_error(DecryptionStatus.INVALID_TOKEN_LIFETIME)
+        return DecryptedToken(DecryptionStatus.INVALID_TOKEN_LIFETIME, id_str, established, site_id, site_key.site_id,
+                          keys.get_identity_scope(), AdvertisingTokenVersion.ADVERTISING_TOKEN_V2, expires)
 
     return DecryptedToken(DecryptionStatus.SUCCESS, id_str, established, site_id, site_key.site_id,
-                          keys.get_identity_scope(), AdvertisingTokenVersion.ADVERTISING_TOKEN_V2)
+                          keys.get_identity_scope(), AdvertisingTokenVersion.ADVERTISING_TOKEN_V2, expires)
 
 
 def _decrypt_token_v3(token_bytes, keys, domain_name, client_type, now, token_version):
@@ -194,7 +197,8 @@ def _decrypt_token_v3(token_bytes, keys, domain_name, client_type, now, token_ve
     expires_ms = int.from_bytes(master_payload[:8], 'big')
     expires = dt.datetime.fromtimestamp(expires_ms / 1000.0, tz=timezone.utc)
     if expires < now:
-        return DecryptedToken.make_error(DecryptionStatus.TOKEN_EXPIRED)
+        return DecryptedToken(DecryptionStatus.TOKEN_EXPIRED, None, None, None, None,
+                          keys.get_identity_scope(), token_version, expires)
 
     # created 8:16
     # operator site id 16:20
@@ -205,7 +209,8 @@ def _decrypt_token_v3(token_bytes, keys, domain_name, client_type, now, token_ve
     site_key_id = int.from_bytes(master_payload[29:33], 'big')
     site_key = keys.get(site_key_id)
     if site_key is None:
-        return DecryptedToken.make_error(DecryptionStatus.NOT_AUTHORIZED_FOR_KEY)
+        return DecryptedToken(DecryptionStatus.NOT_AUTHORIZED_FOR_KEY, None, None, None, site_key.site_id,
+                          keys.get_identity_scope(), token_version, expires)
 
     site_payload = _decrypt_gcm(master_payload[33:], site_key.secret)
 
@@ -217,20 +222,22 @@ def _decrypt_token_v3(token_bytes, keys, domain_name, client_type, now, token_ve
     privacy_bits.frombytes(site_payload[16:20])
 
     if not _is_domain_name_allowed_for_site(client_type, domain_name, privacy_bits):
-        return DecryptedToken.make_error(DecryptionStatus.DOMAIN_NAME_CHECK_FAILED)
+        return DecryptedToken(DecryptionStatus.DOMAIN_NAME_CHECK_FAILED, None, None, site_id, site_key.site_id,
+                          keys.get_identity_scope(), token_version, expires)
 
     established_ms = int.from_bytes(site_payload[20:28], 'big')
     established = dt.datetime.fromtimestamp(established_ms / 1000.0, tz=timezone.utc)
     # refreshed_ms 28:36
 
     if not _token_has_valid_lifetime(keys, client_type, established, expires, now):
-        return DecryptedToken.make_error(DecryptionStatus.INVALID_TOKEN_LIFETIME)
+        return DecryptedToken(DecryptionStatus.INVALID_TOKEN_LIFETIME, None, established, site_id, site_key.site_id,
+                          keys.get_identity_scope(), token_version, expires)
 
     id_bytes = site_payload[36:]
     id_str = base64.b64encode(id_bytes).decode('ascii')
 
     return DecryptedToken(DecryptionStatus.SUCCESS, id_str, established, site_id, site_key.site_id,
-                          keys.get_identity_scope(), token_version)
+                          keys.get_identity_scope(), token_version, expires)
 
 
 def _encrypt_token(uid2, identity_scope, master_key, site_key, site_id, now, token_expiry, ad_token_version):
@@ -518,22 +525,23 @@ class DecryptedToken:
         site_key_site_id (int): site ID of the site key which the token is encrypted with
     """
 
-    def __init__(self, status, uid, established, site_id, site_key_site_id, identity_scope, advertising_token_version):
+    def __init__(self, status, uid, established, site_id, site_key_site_id, identity_scope, advertising_token_version, expiry):
         self.status = status
         self.uid = uid
         self.established = established
         self.site_id = site_id
         self.site_key_site_id = site_key_site_id
         self.identity_scope = identity_scope
         self.advertising_token_version = advertising_token_version
+        self.expiry = expiry
 
     @property
     def success(self):
         return self.status == DecryptionStatus.SUCCESS
 
     @staticmethod
     def make_error(decryption_status):
-        return DecryptedToken(decryption_status, None, None, None, None, None, None)
+        return DecryptedToken(decryption_status, None, None, None, None, None, None, None)
 
 
 class DecryptedData: