Merge pull request #406 from zezha-msft/sas-question-mark

Handle question mark prefix on sas tokens

azure-storage-common/ChangeLog.md

@@ -4,6 +4,7 @@
 
 ## Version XX.XX.XX:
 - Added back the ability to generate account SAS for table service.
+- Fixed bug where a question mark prefix on SAS tokens causes failures.
 
 ## Version 0.37.1:
 - Fixed the return type of __add__ and __or__ methods on the AccountPermissions class

azure-storage-common/azure/storage/common/_auth.py

@@ -94,7 +94,12 @@ class _StorageNoAuthentication(object):
 
 class _StorageSASAuthentication(object):
     def __init__(self, sas_token):
-        self.sas_token = sas_token
+        # ignore ?-prefix (added by tools such as Azure Portal) on sas tokens
+        # doing so avoids double question marks when signing
+        if sas_token[0] == '?':
+            self.sas_token = sas_token[1:]
+        else:
+            self.sas_token = sas_token
 
     def sign_request(self, request):
         # if 'sig=' is present, then the request has already been signed

tests/common/test_account.py

@@ -183,6 +183,37 @@ class StorageAccountTest(StorageTestCase):
         finally:
             service.delete_container(container_name)
 
+    @record
+    def test_account_sas_with_question_mark_prefix(self):
+        # SAS URL is calculated from storage key, so this test runs live only
+        if TestMode.need_recording_file(self.test_mode):
+            return
+
+        # Arrange
+        token = '?' + self.account.generate_shared_access_signature(
+            Services.BLOB,
+            ResourceTypes.OBJECT + ResourceTypes.CONTAINER,
+            AccountPermissions.READ + AccountPermissions.WRITE + AccountPermissions.DELETE + AccountPermissions.CREATE,
+            datetime.utcnow() + timedelta(hours=1),
+        )
+
+        service = BlockBlobService(self.account_name, sas_token=token)
+        data = b'shared access signature with read/write permission on blob'
+        container_name = 'container1'
+        blob_name = 'blob1.txt'
+
+        try:
+            # Act
+            service.create_container(container_name)
+            service.create_blob_from_bytes(container_name, blob_name, data)
+            blob = service.get_blob_to_bytes(container_name, blob_name)
+
+            # Assert
+            self.assertIsNotNone(blob)
+            self.assertEqual(data, blob.content)
+        finally:
+            service.delete_container(container_name)
+
     @record
     def test_generate_account_sas_with_multiple_permissions(self):
         # SAS URL is calculated from storage key, so this test runs live only