Repository: Backblaze/B2_Command_Line_Tool

Issue:
b2.api.B2Api.get_bucket_by_name does not work with bucket-scoped application keys
I am using Duplicity 0.7.17 and b2 1.3.2.

duplicity is executed using

```
duplicity \
    --verbosity debug \
    /backup \
    "b2://$B2_APP_KEY_ID:$B2_APP_KEY@$B2_BUCKET_NAME"
```

Where `$B2_APP_KEY_ID` and `$B2_APP_KEY` are URL-encoded strings which were output from a call to:

```
b2 create-key \
    --bucket "$B2_BUCKET_NAME" \
    "$B2_KEY_NAME" \
    listFiles,readFiles,writeFiles,deleteFiles,listBuckets
```

Duplicity fails with the following traceback:

```
Traceback (innermost last):
  File "/usr/local/bin/duplicity", line 1555, in <module>
    with_tempdir(main)
  File "/usr/local/bin/duplicity", line 1541, in with_tempdir
    fn()
  File "/usr/local/bin/duplicity", line 1380, in main
    action = commandline.ProcessCommandLine(sys.argv[1:])
  File "/usr/local/lib/python2.7/dist-packages/duplicity/commandline.py", line 1135, in ProcessCommandLine
    backup, local_pathname = set_backend(args[0], args[1])
  File "/usr/local/lib/python2.7/dist-packages/duplicity/commandline.py", line 1010, in set_backend
    globals.backend = backend.get_backend(bend)
  File "/usr/local/lib/python2.7/dist-packages/duplicity/backend.py", line 223, in get_backend
    obj = get_backend_object(url_string)
  File "/usr/local/lib/python2.7/dist-packages/duplicity/backend.py", line 209, in get_backend_object
    return factory(pu)
  File "/usr/local/lib/python2.7/dist-packages/duplicity/backends/b2backend.py", line 87, in __init__
    self.bucket = self.service.get_bucket_by_name(bucket_name)
  File "/usr/local/lib/python2.7/dist-packages/b2/api.py", line 222, in get_bucket_by_name
    for bucket in self.list_buckets():
  File "/usr/local/lib/python2.7/dist-packages/logfury/v0_1/trace_call.py", line 84, in wrapper
    return function(*wrapee_args, **wrapee_kwargs)
  File "/usr/local/lib/python2.7/dist-packages/b2/api.py", line 251, in list_buckets
    self.check_bucket_restrictions(bucket_name)
  File "/usr/local/lib/python2.7/dist-packages/logfury/v0_1/trace_call.py", line 84, in wrapper
    return function(*wrapee_args, **wrapee_kwargs)
  File "/usr/local/lib/python2.7/dist-packages/b2/api.py", line 375, in check_bucket_restrictions
    raise RestrictedBucket(allowed_bucket_name)
 RestrictedBucket: Application key is restricted to bucket: pc-backup
```

Internally<sup>[[1]](https://bazaar.launchpad.net/~duplicity-team/duplicity/0.8-series/view/head:/duplicity/backends/b2backend.py#L88)</sup>, Duplicity uses `b2.api.B2Api.get_bucket_by_name`, passing in the name of the bucket. This method calls `b2.api.B2Api.list_buckets` without passing the bucket name, so we get the permission error indicated above.

This looks related to changes in #474.

Discussion:

Patch:
diff --git a/b2/api.py b/b2/api.py
index 017f5ba..a1400e1 100644
--- a/b2/api.py
+++ b/b2/api.py
@@ -205,20 +205,27 @@ class B2Api(object):
 
     def get_bucket_by_name(self, bucket_name):
         """
-        Returns the bucket_id for the given bucket_name.
+        Returns the Bucket for the given bucket_name.
 
-        If we don't already know it from the cache, try fetching it from
-        the B2 service.
+        :param bucket_name: The name of the bucket to return.
+        :return: a Bucket object
+        :raises NonExistentBucket: if the bucket does not exist in the account
         """
-        # If we can get it from the stored info, do that.
+        # Give a useful warning if the current application key does not
+        # allow access to the named bucket.
         self.check_bucket_restrictions(bucket_name)
+
+        # First, try the cache.
         id_ = self.cache.get_bucket_id_or_none_from_bucket_name(bucket_name)
         if id_ is not None:
             return Bucket(self, id_, name=bucket_name)
 
-        for bucket in self.list_buckets():
-            if bucket.name == bucket_name:
-                return bucket
+        # Second, ask the service
+        for bucket in self.list_buckets(bucket_name=bucket_name):
+            assert bucket.name == bucket_name
+            return bucket
+
+        # There is no such bucket.
         raise NonExistentBucket(bucket_name)
 
     def delete_bucket(self, bucket):
@@ -244,25 +251,14 @@ class B2Api(object):
         :param bucket_name: Optional: the name of the one bucket to return.
         :return: A list of Bucket objects.
         """
-        account_id = self.account_info.get_account_id()
+        # Give a useful warning if the current application key does not
+        # allow access to the named bucket.
         self.check_bucket_restrictions(bucket_name)
 
-        # TEMPORARY work around until we fix the API endpoint bug that things requests
-        # with a bucket name are not authorized.  When it's fixed, well just pass the
-        # bucket name (or None) to the raw API.
-        if bucket_name is None:
-            bucket_id = None
-        else:
-            allowed = self.account_info.get_allowed()
-            if allowed['bucketId'] is not None:
-                # We just checked that if there is a bucket restriction we have a bucket name
-                # and it matches.  So if there's a restriction we know that's the bucket we're
-                # looking for.
-                bucket_id = allowed['bucketId']
-            else:
-                bucket_id = self.get_bucket_by_name(bucket_name).id_
+        account_id = self.account_info.get_account_id()
+        self.check_bucket_restrictions(bucket_name)
 
-        response = self.session.list_buckets(account_id, bucket_id=bucket_id)
+        response = self.session.list_buckets(account_id, bucket_name=bucket_name)
         buckets = BucketFactory.from_api_response(self, response)
 
         if bucket_name is not None:
diff --git a/b2/raw_simulator.py b/b2/raw_simulator.py
index 9731370..0fcb999 100644
--- a/b2/raw_simulator.py
+++ b/b2/raw_simulator.py
@@ -767,18 +767,40 @@ class RawSimulator(AbstractRawApi):
         self.file_id_to_bucket_id[response['fileId']] = bucket_id
         return response
 
-    def list_buckets(self, api_url, account_auth_token, account_id, bucket_id=None):
-        self._assert_account_auth(api_url, account_auth_token, account_id, 'listBuckets', bucket_id)
+    def list_buckets(
+        self, api_url, account_auth_token, account_id, bucket_id=None, bucket_name=None
+    ):
+        # First, map the bucket name to a bucket_id, so that we can check auth.
+        if bucket_name is None:
+            bucket_id_for_auth = bucket_id
+        else:
+            bucket_id_for_auth = self._get_bucket_id_or_none_for_bucket_name(bucket_name)
+        self._assert_account_auth(
+            api_url, account_auth_token, account_id, 'listBuckets', bucket_id_for_auth
+        )
+
+        # Do the query
         sorted_buckets = [
-            self.bucket_name_to_bucket[bucket_name]
-            for bucket_name in sorted(six.iterkeys(self.bucket_name_to_bucket))
+            self.bucket_name_to_bucket[name]
+            for name in sorted(six.iterkeys(self.bucket_name_to_bucket))
         ]
         bucket_list = [
             bucket.bucket_dict()
-            for bucket in sorted_buckets if bucket_id is None or bucket.bucket_id == bucket_id
+            for bucket in sorted_buckets if self._bucket_matches(bucket, bucket_id, bucket_name)
         ]
         return dict(buckets=bucket_list)
 
+    def _get_bucket_id_or_none_for_bucket_name(self, bucket_name):
+        for bucket in six.itervalues(self.bucket_name_to_bucket):
+            if bucket.bucket_name == bucket_name:
+                return bucket.bucket_id
+
+    def _bucket_matches(self, bucket, bucket_id, bucket_name):
+        return (
+            (bucket_id is None or bucket.bucket_id == bucket_id) and
+            (bucket_name is None or bucket.bucket_name == bucket_name)
+        )
+
     def list_file_names(
         self, api_url, account_auth, bucket_id, start_file_name=None, max_file_count=None
     ):


Test patch:
diff --git a/test/test_api.py b/test/test_api.py
index adcdb45..f72c336 100644
--- a/test/test_api.py
+++ b/test/test_api.py
@@ -53,6 +53,16 @@ class TestApi(TestBase):
             [b.name for b in self.api.list_buckets(bucket_name=bucket1.name)],
         )
 
+    def test_get_bucket_by_name_with_bucket_restriction(self):
+        self._authorize_account()
+        bucket1 = self.api.create_bucket('bucket1', 'allPrivate')
+        key = self.api.create_key(['listBuckets'], 'key1', bucket_id=bucket1.id_)
+        self.api.authorize_account('production', key['applicationKeyId'], key['applicationKey'])
+        self.assertEqual(
+            bucket1.id_,
+            self.api.get_bucket_by_name('bucket1').id_,
+        )
+
     def test_list_buckets_with_restriction_and_wrong_name(self):
         self._authorize_account()
         bucket1 = self.api.create_bucket('bucket1', 'allPrivate')
@@ -72,4 +82,4 @@ class TestApi(TestBase):
             self.api.list_buckets()
 
     def _authorize_account(self):
-        self.api.authorize_account('production', self.account_id, self.master_key)
\ No newline at end of file
+        self.api.authorize_account('production', self.account_id, self.master_key)