backblaze__b2_command_line_tool-488

The user is reporting a bug in `b2.api.B2Api.get_bucket_by_name`.
When using a bucket-scoped application key, calling `get_bucket_by_name` fails because it calls `list_buckets()` without passing the bucket name, which triggers a `RestrictedBucket` exception.

Looking at the traceback:
```
  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/b2/api.py", line 251, in list_buckets
    self.check_bucket_restrictions(bucket_name)
  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
```

It seems `get_bucket_by_name` should pass the `bucket_name` to `list_buckets(bucket_name=bucket_name)` so that `check_bucket_restrictions` can be satisfied.

Let's start by exploring `b2/api.py`.

{
  "command": "grep -rn --include=\"*.py\" \"get_bucket_by_name\""
}

{
  "status": "ok",
  "stdout": "test/test_console_tool.py:388:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\ntest/test_console_tool.py:395:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\ntest/test_console_tool.py:603:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\ntest/test_console_tool.py:610:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\ntest/test_console_tool.py:794:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\ntest/test_console_tool.py:829:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\ntest/test_console_tool.py:873:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\ntest/test_console_tool.py:913:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\ntest/test_console_tool.py:1070:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\ntest/test_policy.py:69:        api.get_bucket_by_name.return_value = bucket\ntest/test_sync.py:248:        self.api.get_bucket_by_name.return_value = self.bucket\nb2/api.py:206:    def get_bucket_by_name(self, bucket_name):\nb2/api.py:263:                bucket_id = self.get_bucket_by_name(bucket_name).id_\nb2/raw_simulator.py:709:            bucket = self._get_bucket_by_name(bucket_name)\nb2/raw_simulator.py:924:    def _get_bucket_by_name(self, bucket_name):\nb2/sync/folder.py:230:        self.bucket = api.get_bucket_by_name(bucket_name)\nb2/console_tool.py:299:        bucket = self.api.get_bucket_by_name(args.bucketName)\nb2/console_tool.py:405:            bucket_id_or_none = self.api.get_bucket_by_name(args.bucket).id_\nb2/console_tool.py:433:        bucket = self.api.get_bucket_by_name(args.bucketName)\nb2/console_tool.py:529:        bucket = self.api.get_bucket_by_name(args.bucketName)\nb2/console_tool.py:656:        bucket = self.api.get_bucket_by_name(args.bucketName)\nb2/console_tool.py:691:        bucket = self.api.get_bucket_by_name(args.bucketName)\nb2/console_tool.py:733:        bucket = self.api.get_bucket_by_name(args.bucketName)\nb2/console_tool.py:779:        bucket = self.api.get_bucket_by_name(args.bucketName)\nb2/console_tool.py:802:        bucket = self.api.get_bucket_by_name(args.bucketName)\nb2/console_tool.py:932:        bucket = self.api.get_bucket_by_name(args.bucketName)\nb2/console_tool.py:983:        bucket = self.api.get_bucket_by_name(args.bucketName)\nb2/console_tool.py:1236:        bucket = self.api.get_bucket_by_name(args.bucketName)\nb2/console_tool.py:1303:        bucket = self.api.get_bucket_by_name(args.bucketName)\n",
  "stderr": "",
  "exit_code": 0
}

Okay, the main implementation of `get_bucket_by_name` is in `b2/api.py` at line 206.
Let's read that function and the `list_buckets` function.

{
  "end_line": 380,
  "filepath": "b2/api.py",
  "start_line": 200
}

{
  "status": "ok",
  "filepath": "b2/api.py",
  "content": "        )\n        return self.transferer.download_file_from_url(url, download_dest, progress_listener, range_)\n\n    def get_bucket_by_id(self, bucket_id):\n        return Bucket(self, bucket_id)\n\n    def get_bucket_by_name(self, bucket_name):\n        \"\"\"\n        Returns the bucket_id for the given bucket_name.\n\n        If we don't already know it from the cache, try fetching it from\n        the B2 service.\n        \"\"\"\n        # If we can get it from the stored info, do that.\n        self.check_bucket_restrictions(bucket_name)\n        id_ = self.cache.get_bucket_id_or_none_from_bucket_name(bucket_name)\n        if id_ is not None:\n            return Bucket(self, id_, name=bucket_name)\n\n        for bucket in self.list_buckets():\n            if bucket.name == bucket_name:\n                return bucket\n        raise NonExistentBucket(bucket_name)\n\n    def delete_bucket(self, bucket):\n        \"\"\"\n        Deletes the bucket remotely.\n        For legacy reasons it returns whatever server sends in response,\n        but API user should not rely on the response: if it doesn't raise\n        an exception, it means that the operation was a success\n        \"\"\"\n        account_id = self.account_info.get_account_id()\n        return self.session.delete_bucket(account_id, bucket.id_)\n\n    def list_buckets(self, bucket_name=None):\n        \"\"\"\n        Calls b2_list_buckets and returns a list of buckets.\n\n        When no bucket name is specified, returns *all* of the buckets\n        in the account.  When a bucket name is given, returns just that\n        bucket.  When authorized with an application key restricted to\n        one bucket, you must specify the bucket name, or the request\n        will be unauthorized.\n\n        :param bucket_name: Optional: the name of the one bucket to return.\n        :return: A list of Bucket objects.\n        \"\"\"\n        account_id = self.account_info.get_account_id()\n        self.check_bucket_restrictions(bucket_name)\n\n        # TEMPORARY work around until we fix the API endpoint bug that things requests\n        # with a bucket name are not authorized.  When it's fixed, well just pass the\n        # bucket name (or None) to the raw API.\n        if bucket_name is None:\n            bucket_id = None\n        else:\n            allowed = self.account_info.get_allowed()\n            if allowed['bucketId'] is not None:\n                # We just checked that if there is a bucket restriction we have a bucket name\n                # and it matches.  So if there's a restriction we know that's the bucket we're\n                # looking for.\n                bucket_id = allowed['bucketId']\n            else:\n                bucket_id = self.get_bucket_by_name(bucket_name).id_\n\n        response = self.session.list_buckets(account_id, bucket_id=bucket_id)\n        buckets = BucketFactory.from_api_response(self, response)\n\n        if bucket_name is not None:\n            # If a bucket_name is specified we don't clear the cache because the other buckets could still\n            # be valid. So we save the one bucket returned from the list_buckets call.\n            for bucket in buckets:\n                self.cache.save_bucket(bucket)\n        else:\n            # Otherwise we want to clear the cache and save the buckets returned from list_buckets\n            # since we just got a new list of all the buckets for this account.\n            self.cache.set_bucket_name_cache(buckets)\n        return buckets\n\n    def list_parts(self, file_id, start_part_number=None, batch_size=None):\n        \"\"\"\n        Generator that yields a Part for each of the parts that have been uploaded.\n\n        :param file_id: the ID of the large file that is not finished\n        :param start_part_number: the first part number to return.  defaults to the first part.\n        :param batch_size: the number of parts to fetch at a time from the server\n        \"\"\"\n        batch_size = batch_size or 100\n        while True:\n            response = self.session.list_parts(file_id, start_part_number, batch_size)\n            for part_dict in response['parts']:\n                yield PartFactory.from_list_parts_dict(part_dict)\n            start_part_number = response.get('nextPartNumber')\n            if start_part_number is None:\n                break\n\n    # delete/cancel\n    def cancel_large_file(self, file_id):\n        response = self.session.cancel_large_file(file_id)\n        return FileVersionInfoFactory.from_cancel_large_file_response(response)\n\n    def delete_file_version(self, file_id, file_name):\n        # filename argument is not first, because one day it may become optional\n        response = self.session.delete_file_version(file_id, file_name)\n        assert response['fileId'] == file_id\n        assert response['fileName'] == file_name\n        return FileIdAndName(file_id, file_name)\n\n    # download\n    def get_download_url_for_fileid(self, file_id):\n        url = url_for_api(self.account_info, 'b2_download_file_by_id')\n        return '%s?fileId=%s' % (url, file_id)\n\n    def get_download_url_for_file_name(self, bucket_name, file_name):\n        \"\"\"\n        Returns a URL to download the given file by name.\n        \"\"\"\n        self.check_bucket_restrictions(bucket_name)\n        return '%s/file/%s/%s' % (\n            self.account_info.get_download_url(), bucket_name, b2_url_encode(file_name)\n        )\n\n    # keys\n    def create_key(\n        self, capabilities, key_name, valid_duration_seconds=None, bucket_id=None, name_prefix=None\n    ):\n        account_id = self.account_info.get_account_id()\n\n        response = self.session.create_key(\n            account_id,\n            capabilities=capabilities,\n            key_name=key_name,\n            valid_duration_seconds=valid_duration_seconds,\n            bucket_id=bucket_id,\n            name_prefix=name_prefix\n        )\n\n        assert set(response['capabilities']) == set(capabilities)\n        assert response['keyName'] == key_name\n\n        return response\n\n    def delete_key(self, application_key_id):\n\n        response = self.session.delete_key(application_key_id=application_key_id)\n        return response\n\n    def list_keys(self, start_application_key_id=None):\n        account_id = self.account_info.get_account_id()\n",
  "start_line": 200,
  "end_line": 349,
  "total_lines": 372,
  "is_truncated": true
}

Let's look at `get_bucket_by_name`:

```python
    def get_bucket_by_name(self, bucket_name):
        """
        Returns the bucket_id for the given bucket_name.

        If we don't already know it from the cache, try fetching it from
        the B2 service.
        """
        # If we can get it from the stored info, do that.
        self.check_bucket_restrictions(bucket_name)
        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
        raise NonExistentBucket(bucket_name)
```

And `list_buckets`:

```python
    def list_buckets(self, bucket_name=None):
        """
        Calls b2_list_buckets and returns a list of buckets.

        When no bucket name is specified, returns *all* of the buckets
        in the account.  When a bucket name is given, returns just that
        bucket.  When authorized with an application key restricted to
        one bucket, you must specify the bucket name, or the request
        will be unauthorized.

        :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()
        self.check_bucket_restrictions(bucket_name)
        # ...
```

If `get_bucket_by_name` calls `self.list_buckets()` without arguments, then `bucket_name` is `None` in `list_buckets`.
Then `self.check_bucket_restrictions(None)` is called.

Let's find `check_bucket_restrictions` in `b2/api.py`.

{
  "command": "grep -n \"def check_bucket_restrictions\" b2/api.py"
}

{
  "status": "ok",
  "stdout": "359:    def check_bucket_restrictions(self, bucket_name):\n",
  "stderr": "",
  "exit_code": 0
}

Let's read `check_bucket_restrictions`.

{
  "filepath": "b2/api.py",
  "start_line": 359
}

{
  "status": "ok",
  "filepath": "b2/api.py",
  "content": "    def check_bucket_restrictions(self, bucket_name):\n        \"\"\"\n        Checks to see if the allowed field from authorize-account\n        has a bucket restriction.\n\n        If it does, does the bucket_name for a given api call match that.\n        If not it raises a RestrictedBucket error.\n        \"\"\"\n        allowed = self.account_info.get_allowed()\n        allowed_bucket_name = allowed['bucketName']\n\n        if allowed_bucket_name is not None:\n            if allowed_bucket_name != bucket_name:\n                raise RestrictedBucket(allowed_bucket_name)",
  "start_line": 359,
  "end_line": 372,
  "total_lines": 372,
  "is_truncated": false
}

Yes, if `allowed_bucket_name` is not None (i.e., the key is restricted to a bucket) and `bucket_name` is None (because `list_buckets()` was called without arguments), it raises `RestrictedBucket(allowed_bucket_name)`.

The fix is indeed to change `for bucket in self.list_buckets():` to `for bucket in self.list_buckets(bucket_name=bucket_name):` in `get_bucket_by_name`.

Wait, if `self.list_buckets(bucket_name=bucket_name)` is called, it returns a list containing only that bucket if it exists.

Let's check how `list_buckets` is implemented when `bucket_name` is provided.

```python
    def list_buckets(self, bucket_name=None):
        # ...
        account_id = self.account_info.get_account_id()
        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_

        response = self.session.list_buckets(account_id, bucket_id=bucket_id)
        buckets = BucketFactory.from_api_response(self, response)
        # ...
```

Wait! Look at the `else` block in `list_buckets`:
```python
            else:
                bucket_id = self.get_bucket_by_name(bucket_name).id_
```
This is a recursive call!
`get_bucket_by_name` calls `list_buckets(bucket_name)`, which calls `get_bucket_by_name(bucket_name)` if there is no `allowed['bucketId']`.

But `allowed` from `self.account_info.get_allowed()` should have `bucketId` and `bucketName` if it's restricted.

Let's see what `account_info` is. It's in `b2/account_info/`.
The problem states: "b2.api.B2Api.get_bucket_by_name does not work with bucket-scoped application keys".
"RestrictedBucket: Application key is restricted to bucket: pc-backup"

If the application key is bucket-scoped, the `authorize_account` response contains `allowed` field with `bucketName` and `bucketId`.

Let's verify the recursive call.
If `allowed['bucketId']` is NOT None:
`bucket_id = allowed['bucketId']`
`response = self.session.list_buckets(account_id, bucket_id=bucket_id)`
`buckets = BucketFactory.from_api_response(self, response)`
Then `get_bucket_by_name` receives these buckets and iterates through them.

If `allowed['bucketId']` IS None, but the key is restricted to a bucket? That wouldn't make sense according to B2 API. If it's restricted, it has both `bucketId` and `bucketName`.

Wait, what if the account info doesn't have `bucketId` yet?

Let's trace `get_bucket_by_name(bucket_name)`:
1. `self.check_bucket_restrictions(bucket_name)`:
   - If key restricted to `B`, and `bucket_name` is `B`, OK.
   - If key restricted to `B`, and `bucket_name` is `C`, raises `RestrictedBucket`.
   - If key not restricted, OK.
2. Cache check.
3. `for bucket in self.list_buckets():`
   - `list_buckets(None)` is called.
   - `self.check_bucket_restrictions(None)`:
     - If key restricted to `B`, and `bucket_name` is `None`, raises `RestrictedBucket`. **BINGO**.

So changing `self.list_buckets()` to `self.list_buckets(bucket_name=bucket_name)` solves the problem for restricted keys.

If I change it to `self.list_buckets(bucket_name=bucket_name)`, then in `list_buckets`:
1. `self.check_bucket_restrictions(bucket_name)`: OK.
2. `if bucket_name is None:` (False)
3. `else:`
   - `allowed = self.account_info.get_allowed()`
   - `if allowed['bucketId'] is not None:`
     - `bucket_id = allowed['bucketId']`
     - `response = self.session.list_buckets(account_id, bucket_id=bucket_id)`
     - `buckets = ...`
     - `return buckets` (one bucket)
   - `else:`
     - `bucket_id = self.get_bucket_by_name(bucket_name).id_`
     - THIS IS still a recursive call!

Wait, if `allowed['bucketId']` is None, does that mean it's NOT restricted?
If it's NOT restricted, then `allowed = {}` or `allowed = {'bucketId': None, 'bucketName': None}`.

If not restricted:
`list_buckets(bucket_name='my-bucket')`
`allowed['bucketId']` is None.
It calls `self.get_bucket_by_name('my-bucket').id_`.
And `get_bucket_by_name('my-bucket')` calls `list_buckets(None)`.
`list_buckets(None)` doesn't call `get_bucket_by_name`.
It calls `self.session.list_buckets(account_id, bucket_id=None)`.
This returns all buckets.
Then `get_bucket_by_name` finds the bucket and returns it.

So it works for non-restricted keys.
For restricted keys, `allowed['bucketId']` should be present.

The recursive call `bucket_id = self.get_bucket_by_name(bucket_name).id_` only happens if `bucket_name` is provided but `allowed['bucketId']` is NOT provided.
Is this possible?
If the key is not restricted, `allowed['bucketId']` is None.
If the key IS restricted, `allowed['bucketId']` should be set.

So for a restricted key, the recursive call is avoided.
For a non-restricted key, it's:
`get_bucket_by_name(name)` -> `list_buckets(name)` -> `get_bucket_by_name(name)` -> `list_buckets(None)` -> `session.list_buckets()` -> `get_bucket_by_name` returns it.

Wait, if `get_bucket_by_name(name)` calls `list_buckets(name)`, and `list_buckets(name)` calls `get_bucket_by_name(name)`, we have infinite recursion!

Let's check:
```python
    def get_bucket_by_name(self, bucket_name):
        # ...
        for bucket in self.list_buckets():
            if bucket.name == bucket_name:
                return bucket
        raise NonExistentBucket(bucket_name)

    def list_buckets(self, bucket_name=None):
        # ...
        if bucket_name is None:
            bucket_id = None
        else:
            allowed = self.account_info.get_allowed()
            if allowed['bucketId'] is not None:
                bucket_id = allowed['bucketId']
            else:
                bucket_id = self.get_bucket_by_name(bucket_name).id_
        # ...
```

If I change `get_bucket_by_name` to call `self.list_buckets(bucket_name)`, then:
`get_bucket_by_name('B')` calls `list_buckets('B')`
`list_buckets('B')`, since not restricted, calls `get_bucket_by_name('B')`
CRASH.

So I cannot simply change `self.list_buckets()` to `self.list_buckets(bucket_name=bucket_name)` if the `list_buckets` implementation is as it is.

I should probably change `list_buckets` to NOT call `get_bucket_by_name`, or change `get_bucket_by_name` to use `session.list_buckets` or something.

But the goal of `list_buckets(bucket_name=bucket_name)` is to limit the request.

Let's look at `list_buckets` again:
```python
    def list_buckets(self, bucket_name=None):
        """
        Calls b2_list_buckets and returns a list of buckets.

        When no bucket name is specified, returns *all* of the buckets
        in the account.  When a bucket name is given, returns just that
        bucket.  When authorized with an application key restricted to
        one bucket, you must specify the bucket name, or the request
        will be unauthorized.

        :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()
        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_

        response = self.session.list_buckets(account_id, bucket_id=bucket_id)
        buckets = BucketFactory.from_api_response(self, response)
        # ...
```

The "TEMPORARY work around" seems to be the problem.
The B2 API for `b2_list_buckets` takes an optional `bucketId`.
If `bucketId` is provided, it returns only that bucket.
If not, it returns all buckets.

The issue here is related to how application keys work.
If an application key is restricted to a bucket, calling `b2_list_buckets` without `bucketId` returns 403 Forbidden.

So, if we have a restricted key, we MUST provide `bucketId`.
`allowed['bucketId']` provides that.

The recursive call `self.get_bucket_by_name(bucket_name).id_` is used when we want to list a specific bucket by name, but we don't have its ID yet, AND we are NOT restricted to a single bucket.

But if we are NOT restricted, we can just call `self.session.list_buckets(account_id)` and find the bucket in the result.

Wait, `get_bucket_by_name` does:
```python
        for bucket in self.list_buckets():
            if bucket.name == bucket_name:
                return bucket
```
This calls `list_buckets(None)`, which calls `self.session.list_buckets(account_id, bucket_id=None)`.
This is perfectly fine for non-restricted keys.
For restricted keys, this fails.

If we change `get_bucket_by_name` to:
```python
        for bucket in self.list_buckets(bucket_name=bucket_name):
            if bucket.name == bucket_name:
                return bucket
```

Then for restricted keys:
`list_buckets('B')` -> `allowed['bucketId']` is set -> `session.list_buckets(account_id, bucket_id=allowed['bucketId'])` -> returns bucket 'B' -> `get_bucket_by_name` returns bucket 'B'. OK.

For non-restricted keys:
`get_bucket_by_name('B')` -> `list_buckets('B')` -> `allowed['bucketId']` is None -> `get_bucket_by_name('B')` -> INFINITE RECURSION.

So, the recursive call in `list_buckets` must be removed or handled.

If I change `list_buckets`'s recursive call to instead use `list_buckets(None)` and filter?
But `list_buckets` is supposed to return a list of buckets.

If `bucket_name` is provided and `allowed['bucketId']` is None, it means we are not restricted.
We can just call `self.session.list_buckets(account_id, bucket_id=None)`, and then filter the result to only include `bucket_name`.

Let's see `list_buckets` again.

```python
    def list_buckets(self, bucket_name=None):
        # ...
        if bucket_name is None:
            bucket_id = None
        else:
            allowed = self.account_info.get_allowed()
            if allowed['bucketId'] is not None:
                bucket_id = allowed['bucketId']
            else:
                bucket_id = self.get_bucket_by_name(bucket_name).id_

        response = self.session.list_buckets(account_id, bucket_id=bucket_id)
        buckets = BucketFactory.from_api_response(self, response)
        # ...
```

If I change the `else` to:
```python
            else:
                bucket_id = None
```
Then `self.session.list_buckets(account_id, bucket_id=None)` is called.
`buckets` will contain all buckets.
Then we can filter them if `bucket_name` is not None.

Wait, `list_buckets` currently doesn't filter the `buckets` list if `bucket_name` is not None, it just uses `bucket_id` to let the server filter.

If `allowed['bucketId']` is None, we can't use the server to filter by `bucketId` unless we already know the `bucketId`.
If we don't know the `bucketId`, we must list all and filter.

Let's see how `get_bucket_by_name` is used. It's and a way to get a `Bucket` object from a name.

If I change `list_buckets` to:

```python
    def list_buckets(self, bucket_name=None):
        # ...
        account_id = self.account_info.get_account_id()
        self.check_bucket_restrictions(bucket_name)

        if bucket_name is None:
            bucket_id = None
        else:
            allowed = self.account_info.get_allowed()
            if allowed['bucketId'] is not None:
                bucket_id = allowed['bucketId']
            else:
                bucket_id = None # list all and filter later

        response = self.session.list_buckets(account_id, bucket_id=bucket_id)
        buckets = BucketFactory.from_api_response(self, response)

        if bucket_name is not None:
            buckets = [b for b in buckets if b.name == bucket_name]
            # ... cache saving ...
```

Wait, if I do this, then `get_bucket_by_name` can call `list_buckets(bucket_name=bucket_name)`.

Let's refine the proposed changes.

In `get_bucket_by_name`:
```python
-        for bucket in self.list_buckets():
+        for bucket in self.list_buckets(bucket_name=bucket_name):
```

In `list_buckets`:
```python
-            else:
-                bucket_id = self.get_bucket_by_name(bucket_name).id_
+            else:
+                bucket_id = None
```
And add filtering if `bucket_name` is not None.

Wait, `list_buckets` already has some complex cache logic at the end:
```python
        if bucket_name is not None:
            # If a bucket_name is specified we don't clear the cache because the other buckets could still
            # be valid. So we save the one bucket returned from the list_buckets call.
            for bucket in buckets:
                self.cache.save_bucket(bucket)
        else:
            # Otherwise we want to clear the cache and save the buckets returned from list_buckets
            # since we just got a new list of all the buckets for this account.
            self.cache.set_bucket_name_cache(buckets)
        return buckets
```

If I change `bucket_id = None` for non-restricted keys, `buckets` will contain all buckets.
If I then return `buckets` as is, `list_buckets(bucket_name='B')` will return ALL buckets instead of just 'B'. This is a breaking change if other people rely on `list_buckets(bucket_name='B')` returning

{
  "command": "grep -rn \"list_buckets(\" b2/api.py"
}

{
  "status": "ok",
  "stdout": "134:                allowed_bucket_response = self.raw_api.list_buckets(\n219:        for bucket in self.list_buckets():\n234:    def list_buckets(self, bucket_name=None):\n265:        response = self.session.list_buckets(account_id, bucket_id=bucket_id)\n",
  "stderr": "",
  "exit_code": 0
}

The only call to `list_buckets` in `b2/api.py` (besides the definition) is in `get_bucket_by_name` (line 219, which is `for bucket in self.list_buckets():`).

Wait, let's check other files.
`grep -rn "list_buckets(" .` (excluding `b2/api.py` if needed, but let's just run it)

{
  "command": "grep -rn \"list_buckets(\" ."
}

{
  "status": "ok",
  "stdout": "./test/test_console_tool.py:105:    def test_authorize_key_without_list_buckets(self):\n./test/test_console_tool.py:298:        self.b2_api.list_buckets('my-bucket-a')\n./test/test_api.py:27:    def test_list_buckets(self):\n./test/test_api.py:33:            [b.name for b in self.api.list_buckets()],\n./test/test_api.py:42:            [b.name for b in self.api.list_buckets(bucket_name='bucket1')],\n./test/test_api.py:53:            [b.name for b in self.api.list_buckets(bucket_name=bucket1.name)],\n./test/test_api.py:63:            self.api.list_buckets(bucket_name=bucket2.name)\n./test/test_api.py:72:            self.api.list_buckets()\n./b2/api.py:134:                allowed_bucket_response = self.raw_api.list_buckets(\n./b2/api.py:219:        for bucket in self.list_buckets():\n./b2/api.py:234:    def list_buckets(self, bucket_name=None):\n./b2/api.py:265:        response = self.session.list_buckets(account_id, bucket_id=bucket_id)\n./b2/raw_simulator.py:770:    def list_buckets(self, api_url, account_auth_token, account_id, bucket_id=None):\n./b2/console_tool.py:587:        for b in self.api.list_buckets(args.bucketName):\n./b2/console_tool.py:755:        for b in self.api.list_buckets():\n./b2/console_tool.py:883:            self.bucket_id_to_bucket_name = dict((b.id_, b.name) for b in self.api.list_buckets())\n./b2/raw_api.py:287:    def list_buckets(\n./b2/raw_api.py:600:    bucket_list_dict = raw_api.list_buckets(api_url, account_auth_token, account_id)\n",
  "stderr": "",
  "exit_code": 0
}

In `test/test_api.py`:
```python
./test/test_api.py:42:            [b.name for b in self.api.list_buckets(bucket_name='bucket1')],
./test/test_api.py:53:            [b.name for b in self.api.list_buckets(bucket_name=bucket1.name)],
./test/test_api.py:63:            self.api.list_buckets(bucket_name=bucket2.name)
```
These tests expect `list_buckets(bucket_name=...)` to return only the requested bucket.

And `b2/console_tool.py`:
```python
./b2/console_tool.py:587:        for b in self.api.list_buckets(args.bucketName):
```
If `args.bucketName` is provided, it probably expects only that bucket.

So I MUST filter the results in `list_buckets` if `bucket_name` is provided but `bucket_id` was None.

Let's implement this.

First, let's create a reproduction test case to confirm the bug.
I'll look at `test/test_api.py` for inspiration.

{
  "filepath": "test/test_api.py"
}

{
  "status": "ok",
  "filepath": "test/test_api.py",
  "content": "######################################################################\n#\n# File: test/test_api.py\n#\n# Copyright 2018 Backblaze Inc. All Rights Reserved.\n#\n# License https://www.backblaze.com/using_b2_code.html\n#\n######################################################################\n\nfrom .test_base import TestBase\nfrom b2.account_info import InMemoryAccountInfo\nfrom b2.api import B2Api\nfrom b2.cache import DummyCache\nfrom b2.exception import RestrictedBucket\nfrom b2.raw_simulator import RawSimulator\n\n\nclass TestApi(TestBase):\n    def setUp(self):\n        self.account_info = InMemoryAccountInfo()\n        self.cache = DummyCache()\n        self.raw_api = RawSimulator()\n        self.api = B2Api(self.account_info, self.cache, self.raw_api)\n        (self.account_id, self.master_key) = self.raw_api.create_account()\n\n    def test_list_buckets(self):\n        self._authorize_account()\n        self.api.create_bucket('bucket1', 'allPrivate')\n        self.api.create_bucket('bucket2', 'allPrivate')\n        self.assertEqual(\n            ['bucket1', 'bucket2'],\n            [b.name for b in self.api.list_buckets()],\n        )\n\n    def test_list_buckets_with_name(self):\n        self._authorize_account()\n        self.api.create_bucket('bucket1', 'allPrivate')\n        self.api.create_bucket('bucket2', 'allPrivate')\n        self.assertEqual(\n            ['bucket1'],\n            [b.name for b in self.api.list_buckets(bucket_name='bucket1')],\n        )\n\n    def test_list_buckets_with_restriction(self):\n        self._authorize_account()\n        bucket1 = self.api.create_bucket('bucket1', 'allPrivate')\n        self.api.create_bucket('bucket2', 'allPrivate')\n        key = self.api.create_key(['listBuckets'], 'key1', bucket_id=bucket1.id_)\n        self.api.authorize_account('production', key['applicationKeyId'], key['applicationKey'])\n        self.assertEqual(\n            ['bucket1'],\n            [b.name for b in self.api.list_buckets(bucket_name=bucket1.name)],\n        )\n\n    def test_list_buckets_with_restriction_and_wrong_name(self):\n        self._authorize_account()\n        bucket1 = self.api.create_bucket('bucket1', 'allPrivate')\n        bucket2 = self.api.create_bucket('bucket2', 'allPrivate')\n        key = self.api.create_key(['listBuckets'], 'key1', bucket_id=bucket1.id_)\n        self.api.authorize_account('production', key['applicationKeyId'], key['applicationKey'])\n        with self.assertRaises(RestrictedBucket):\n            self.api.list_buckets(bucket_name=bucket2.name)\n\n    def test_list_buckets_with_restriction_and_no_name(self):\n        self._authorize_account()\n        bucket1 = self.api.create_bucket('bucket1', 'allPrivate')\n        self.api.create_bucket('bucket2', 'allPrivate')\n        key = self.api.create_key(['listBuckets'], 'key1', bucket_id=bucket1.id_)\n        self.api.authorize_account('production', key['applicationKeyId'], key['applicationKey'])\n        with self.assertRaises(RestrictedBucket):\n            self.api.list_buckets()\n\n    def _authorize_account(self):\n        self.api.authorize_account('production', self.account_id, self.master_key)",
  "start_line": 1,
  "end_line": 75,
  "total_lines": 75,
  "is_truncated": false
}

I can add a test case to `TestApi` in `test/test_api.py` that calls `get_bucket_by_name` with a restricted key.

```python
    def test_get_bucket_by_name_with_restriction(self):
        self._authorize_account()
        bucket1 = self.api.create_bucket('bucket1', 'allPrivate')
        self.api.create_bucket('bucket2', 'allPrivate')
        key = self.api.create_key(['listBuckets'], 'key1', bucket_id=bucket1.id_)
        self.api.authorize_account('production', key['applicationKeyId'], key['applicationKey'])
        bucket = self.api.get_bucket_by_name(bucket1.name)
        self.assertEqual(bucket.id_, bucket1.id_)
```

Wait, `create_key` in `B2Api` calls `session.create_key`.
Let's see `B2Api.create_key`'s parameters.
Looking at `b2/api.py` read earlier:
```python
    def create_key(
        self, capabilities, key_name, valid_duration_seconds=None, bucket_id=None, name_prefix=None
    ):
        account_id = self.account_info.get_account_id()

        response = self.session.create_key(
            account_id,
            capabilities=capabilities,
            key_name=key_name,
            valid_duration_seconds=valid_duration_seconds,
            bucket_id=bucket_id,
            name_prefix=name_prefix
        )
```

The reproduction test case:
```python
        key = self.api.create_key(['listBuckets'], 'key1', bucket_id=bucket1.id_)
```
This should create a key restricted to `bucket1`.
Then `self.api.authorize_account` will update `account_info` with the restricted info.

Let's run this reproduction in a separate file.

{
  "command": "python3 - <<'PY'\nfrom b2.account_info import InMemoryAccountInfo\nfrom b2.api import B2Api\nfrom b2.cache import DummyCache\nfrom b2.exception import RestrictedBucket\nfrom b2.raw_simulator import RawSimulator\n\nclass TestRepro:\n    def setUp(self):\n        self.account_info = InMemoryAccountInfo()\n        self.cache = DummyCache()\n        self.raw_api = RawSimulator()\n        self.api = B2Api(self.account_info, self.cache, self.raw_api)\n        (self.account_id, self.master_key) = self.raw_api.create_account()\n        self.api.authorize_account('production', self.account_id, self.master_key)\n\n    def reproduce(self):\n        self.setUp()\n        bucket1 = self.api.create_bucket('bucket1', 'allPrivate')\n        self.api.create_bucket('bucket2', 'allPrivate')\n        # Create a key restricted to bucket1\n        key = self.api.create_key(['listBuckets'], 'key1', bucket_id=bucket1.id_)\n        # Use this key to authorize\n        self.api.authorize_account('production', key['applicationKeyId'], key['applicationKey'])\n        \n        print(\"Calling get_bucket_by_name...\")\n        try:\n            bucket = self.api.get_bucket_by_name('bucket1')\n            print(\"Success!\")\n            print(f\"Bucket ID: {bucket.id_}\")\n        except Exception as e:\n            print(f\"Failed with: {type(e).__name__}: {e}\")\n\nrepro = TestRepro()\nrepro.reproduce()\nPY\n"
}

{
  "status": "ok",
  "stdout": "Calling get_bucket_by_name...\nFailed with: RestrictedBucket: Application key is restricted to bucket: bucket1\n",
  "stderr": "",
  "exit_code": 0
}