backblaze__b2_command_line_tool-488

The user reports an issue where `b2.api.B2Api.get_bucket_by_name` doesn't work with bucket-scoped application keys.
The traceback shows:
`b2/api.py`, line 222, in `get_bucket_by_name`
  `for bucket in self.list_buckets():`
`b2/api.py`, line 251, in `list_buckets`
  `self.check_bucket_restrictions(bucket_name)`
`b2/api.py`, line 375, in `check_bucket_restrictions`
  `raise RestrictedBucket(allowed_bucket_name)`
`RestrictedBucket: Application key is restricted to bucket: pc-backup`

The issue seems to be that `get_bucket_by_name(bucket_name)` calls `list_buckets()` without passing the `bucket_name`, which in turn triggers a `RestrictedBucket` exception if the application key is restricted to a specific bucket.

If the key is restricted to `pc-backup` and we call `list_buckets()`, it might be trying to list all buckets, but the key only has permission for one specific bucket. However, the `list_buckets` method might accept an optional `bucket_name` to filter or specifically request a bucket.

Let's examine `b2/api.py`.

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

{
  "status": "ok",
  "stdout": "./b2/api.py:206:    def get_bucket_by_name(self, bucket_name):\n",
  "stderr": "",
  "exit_code": 0
}

Okay, `get_bucket_by_name` is in `b2/api.py` at line 206.
Let's examine `b2/api.py` around lines 206, 251 (list_buckets), and 375 (check_bucket_restrictions).

{
  "end_line": 400,
  "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
}

I need to see `check_bucket_restrictions` as well, which is after line 349.

{
  "end_line": 400,
  "filepath": "b2/api.py",
  "start_line": 350
}

{
  "status": "ok",
  "filepath": "b2/api.py",
  "content": "        return self.session.list_keys(\n            account_id, max_key_count=1000, start_application_key_id=start_application_key_id\n        )\n\n    # other\n    def get_file_info(self, file_id):\n        \"\"\" legacy interface which just returns whatever remote API returns \"\"\"\n        return self.session.get_file_info(file_id)\n\n    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": 350,
  "end_line": 372,
  "total_lines": 372,
  "is_truncated": false
}

Okay, I see the issue.

```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)
```

In `get_bucket_by_name`, it calls `self.list_buckets()` without any arguments.
Now let's look at `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)
        # ...
```

And `check_bucket_restrictions`:

```python
    def check_bucket_restrictions(self, bucket_name):
        """
        Checks to see if the allowed field from authorize-account
        has a bucket restriction.

        If it does, does the bucket_name for a given api call match that.
        If not it raises a RestrictedBucket error.
        """
        allowed = self.account_info.get_allowed()
        allowed_bucket_name = allowed['bucketName']

        if allowed_bucket_name is not None:
            if allowed_bucket_name != bucket_name:
                raise RestrictedBucket(allowed_bucket_name)
```

If `bucket_name` is `None` (which it is when `list_buckets()` is called from `get_bucket_by_name`), and `allowed_bucket_name` is not `None` (bucket-scoped key), then `allowed_bucket_name != None` is True, and it raises `RestrictedBucket`.

So in `get_bucket_by_name(self, bucket_name)`, it should call `self.list_buckets(bucket_name)`.

Wait, if `list_buckets(bucket_name)` is called, what happens?

```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)
        # ...
```

If `list_buckets(bucket_name)` is called, and `bucket_name` is provided, but there is NO bucket restriction (`allowed['bucketId']` is `None`), it calls `self.get_bucket_by_name(bucket_name).id_`.
This creates a recursion!
`get_bucket_by_name` -> `list_buckets` -> `get_bucket_by_name`.

But if there IS a bucket restriction, `allowed['bucketId']` will be set, and it will use that `bucket_id` to call `self.session.list_buckets(account_id, bucket_id=bucket_id)`, avoiding the recursion.

The problem is specifically when we have a bucket-scoped key. In that case, `get_bucket_by_name` should call `list_buckets(bucket_name)`.

If we don't have a bucket-scoped key, `get_bucket_by_name` currently calls `list_buckets()` (which calls `list_buckets(None)`), which returns all buckets.

If we change `get_bucket_by_name` to call `list_buckets(bucket_name)`, then for non-restricted keys:
1. `get_bucket_by_name(bucket_name)` calls `list_buckets(bucket_name)`.
2. `list_buckets(bucket_name)` sees `bucket_name` is not None.
3. `list_buckets` sees `allowed['bucketId']` is None.
4. `list_buckets` calls `self.get_bucket_by_name(bucket_name).id_` -> Recursion.

So we can't just change it to `list_buckets(bucket_name)` unconditionally.

However, in `get_bucket_by_name`, we already verified the bucket name against restrictions:
```python
    def get_bucket_by_name(self, bucket_name):
        # ...
        self.check_bucket_restrictions(bucket_name)
        # ...
        for bucket in self.list_buckets():
            if bucket.name == bucket_name:
                return bucket
        raise NonExistentBucket(bucket_name)
```

If `check_bucket_restrictions(bucket_name)` passed, and we have a bucket restriction, it means `bucket_name` MUST be the restricted bucket.

If we have a bucket restriction, `self.list_buckets()` fails.
If we have a bucket restriction, `self.list_buckets(bucket_name)` succeeds.

If we DON'T have a bucket restriction, `self.list_buckets()` succeeds.
If we DON'T have a bucket restriction, `self.list_buckets(bucket_name)` recurses.

So `get_bucket_by_name` should check if there is a bucket restriction before calling `list_buckets`.

```python
    def get_bucket_by_name(self, bucket_name):
        # ...
        self.check_bucket_restrictions(bucket_name)
        # ...
        
        # Check if there's a restriction. If so, we MUST pass bucket_name to list_buckets.
        allowed = self.account_info.get_allowed()
        if allowed['bucketName'] is not None:
            for bucket in self.list_buckets(bucket_name):
                if bucket.name == bucket_name:
                    return bucket
        else:
            for bucket in self.list_buckets():
                if bucket.name == bucket_name:
                    return bucket
        raise NonExistentBucket(bucket_name)
```

Wait, `list_buckets(bucket_name)` returns a list of buckets. If there's a restriction, it returns just that bucket.

Let's see if I can reproduce this. I need a mock for `account_info` and `cache` and `session`.
The traceback shows the error comes from `check_bucket_restrictions`.

I can write a small reproduction script.

{
  "command": "python3 - <<'PY'\nimport sys\nfrom unittest.mock import MagicMock\nfrom b2.api import B2Api, RestrictedBucket\n\nclass MockAccountInfo:\n    def __init__(self, allowed_bucket_name=None, allowed_bucket_id=None):\n        self.allowed = {'bucketName': allowed_bucket_name, 'bucketId': allowed_bucket_id}\n    def get_account_id(self):\n        return 'account_id'\n    def get_allowed(self):\n        return self.allowed\n\nclass MockCache:\n    def get_bucket_id_or_none_from_bucket_name(self, name):\n        return None\n    def save_bucket(self, bucket):\n        pass\n    def set_bucket_name_cache(self, buckets):\n        pass\n\ndef test_bucket_scoped_key():\n    # Setup a bucket-scoped key\n    account_info = MockAccountInfo(allowed_bucket_name='my-bucket', allowed_bucket_id='my-bucket-id')\n    cache = MockCache()\n    session = MagicMock()\n    \n    # B2Api needs account_info, cache, session, transferer\n    # We can mock transferer too\n    api = B2Api(account_info, cache, session, MagicMock())\n    \n    try:\n        api.get_bucket_by_name('my-bucket')\n        print(\"SUCCESS\")\n    except RestrictedBucket as e:\n        print(f\"FAILED: Caught expected RestrictedBucket: {e}\")\n    except Exception as e:\n        print(f\"FAILED: Caught unexpected exception: {type(e).__name__}: {e}\")\n\ntest_bucket_scoped_key()\nPY\n"
}

{
  "status": "ok",
  "stdout": "FAILED: Caught expected RestrictedBucket: Application key is restricted to bucket: my-bucket\n",
  "stderr": "",
  "exit_code": 0
}

Wait, my reproduction script printed "FAILED: Caught expected RestrictedBucket", but "expected" means it's the bug I'm trying to fix. If it catches `RestrictedBucket`, it's failing.

Let's re-read my script.
```python
    try:
        api.get_bucket_by_name('my-bucket')
        print("SUCCESS")
    except RestrictedBucket as e:
        print(f"FAILED: Caught expected RestrictedBucket: {e}")
```
Yes, it failed because it caught `RestrictedBucket`. This reproduces the bug.

Now let's verify the fix. I'll modify the code in the reproduction script to simulate the fix.

Actually, I'll just apply the fix to `b2/api.py` and then run the reproduction script again.

Wait, I should be careful about the recursion I mentioned earlier.
If `allowed['bucketName']` is None, `get_bucket_by_name` calls `list_buckets()`.
If `allowed['bucketName']` is NOT None, `get_bucket_by_name` calls `list_buckets(bucket_name)`.

In `list_buckets(bucket_name)`:
If `bucket_name` is NOT None:
  If `allowed['bucketId']` is NOT None:
    `bucket_id = allowed['bucketId']`
    `response = self.session.list_buckets(account_id, bucket_id=bucket_id)`
    NO RECURSION.
  Else:
    `bucket_id = self.get_bucket_by_name(bucket_name).id_`
    RECURSION.

So if `allowed['bucketName']` is not None, does it mean `allowed['bucketId']` is also not None?
Let's check how `allowed` is structured. It seems to come from `account_info`.

In `B2Api.list_buckets`:
```python
        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 we have a bucket-scoped key, both `bucketName` and `bucketId` should be present in `allowed`.

Let's check if there's any case where `bucketName` is present but `bucketId` is not.
Usually, when an application key is created for a specific bucket, the B2 API returns both the name and the ID of that bucket in the `authorized` response.

If I change `get_bucket_by_name` to:

```python
    def get_bucket_by_name(self, bucket_name):
        # ...
        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)

        allowed = self.account_info.get_allowed()
        if allowed['bucketName'] is not None:
            for bucket in self.list_buckets(bucket_name):
                if bucket.name == bucket_name:
                    return bucket
        else:
            for bucket in self.list_buckets():
                if bucket.name == bucket_name:
                    return bucket
        raise NonExistentBucket(bucket_name)
```

This should work as long as `allowed['bucketName'] is not None` implies `allowed['bucketId'] is not None`.
If `allowed['bucketName']` is not None, but `allowed['bucketId']` is None, then `list_buckets(bucket_name)` will still recurse.

But `check_bucket_restrictions` uses `allowed['bucketName']`.

```python
    def check_bucket_restrictions(self, bucket_name):
        # ...
        allowed = self.account_info.get_allowed()
        allowed_bucket_name = allowed['bucketName']

        if allowed_bucket_name is not None:
            if allowed_bucket_name != bucket_name:
                raise RestrictedBucket(allowed_bucket_name)
```

If I have a bucket-scoped key, `allowed['bucketName']` will be set.

Is there any other way to get the bucket ID without recursing if we have the name?
`self.cache.get_bucket_id_or_none_from_bucket_name(bucket_name)` is already tried.

Actually, `list_buckets(bucket_name)`'s recursion is specifically for the case where we want a specific bucket but don't have a restriction but we also don't have its ID.

If we HAVE a restriction to `bucket_name`, then `allowed['bucketId']` SHOULD be set.

Let's look at `b2/account_info/in_memory.py` (if it exists) to see how `allowed` is populated.

{
  "command": "grep -rn \"get_allowed\" b2/account_info/"
}

{
  "status": "ok",
  "stdout": "b2/account_info/abstract.py:118:    def get_allowed(self):\nb2/account_info/sqlite_account_info.py:255:    def get_allowed(self):\nb2/account_info/in_memory.py:112:    def get_allowed(self):\n",
  "stderr": "grep: b2/account_info/__pycache__/abstract.cpython-37.pyc: binary file matches\ngrep: b2/account_info/__pycache__/in_memory.cpython-37.pyc: binary file matches\ngrep: b2/account_info/__pycache__/sqlite_account_info.cpython-37.pyc: binary file matches\n",
  "exit_code": 0
}

Let's check `b2/account_info/in_memory.py`.

{
  "end_line": 130,
  "filepath": "b2/account_info/in_memory.py",
  "start_line": 100
}

{
  "status": "ok",
  "filepath": "b2/account_info/in_memory.py",
  "content": "    def get_download_url(self):\n        return self._download_url\n\n    @_raise_missing_if_result_is_none\n    def get_minimum_part_size(self):\n        return self._minimum_part_size\n\n    @_raise_missing_if_result_is_none\n    def get_realm(self):\n        return self._realm\n\n    @_raise_missing_if_result_is_none\n    def get_allowed(self):\n        return self._allowed",
  "start_line": 100,
  "end_line": 113,
  "total_lines": 113,
  "is_truncated": false
}

It just returns `self._allowed`.

Let's look at `b2/api.py` again.
Maybe I can just pass `bucket_name` to `list_buckets()` and handle the recursion? No, that's not good.

If I change `get_bucket_by_name` to call `list_buckets(bucket_name)` if `allowed['bucketName']` is set, I should be safe because if `allowed['bucketName']` is set, `allowed['bucketId']` must also be set.

Let's verify this hypothesis. In B2, if you create an application key restricted to a bucket, you get both the bucket name and bucket ID in the authorization response.

If `allowed['bucketName']` is not None, then `allowed['bucketId']` should not be None.

Let's double check `B2Api.list_buckets` again.

```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)
        # ...
```

Wait! If `bucket_name` is provided, and `allowed['bucketId']` is `None`, it calls `self.get_bucket_by_name(bucket_name).id_`.
And `get_bucket_by_name(bucket_name)` calls `list_buckets()` (with `None`).

So:
`get_bucket_by_name('foo')` -> `list_buckets()` -> `session.list_buckets(account_id, bucket_id=None)` -> returns all buckets.
This works for non-restricted keys.

For restricted keys:
`get_bucket_by_name('foo')` -> `list_buckets()` -> `check_bucket_restrictions(None)` -> RAISES `RestrictedBucket`.

So if we have a restricted key, `get_bucket_by_name('foo')` MUST call `list_buckets('foo')`.
And `list_buckets('foo')` for a restricted key:
`check_bucket_restrictions('foo')` -> PASS.
`allowed['bucketId']` is NOT None -> `bucket_id = allowed['bucketId']`.
`session.list_buckets(account_id, bucket_id=bucket_id)` -> WORKS.

So the fix is indeed: In `get_bucket_by_name`, if `allowed['bucketName']` is not None, call `list_buckets(bucket_name)`.

Let's refine the implementation in `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)

        allowed = self.account_info.get_allowed()
        if allowed['bucketName'] is not None:
            # We have a bucket restriction, so we must pass the bucket name to list_buckets.
            for bucket in self.list_buckets(bucket_name):
                if bucket.name == bucket_name:
                    return bucket
        else:
            for bucket in self.list_buckets():
                if bucket.name == bucket_name:
                    return bucket
        raise NonExistentBucket(bucket_name)
```

Wait, if `allowed['bucketName']` is set, then `list_buckets(bucket_name)` will return only that bucket. The loop is still fine.

Is there any case where `allowed['bucketName']` is set but `allowed['bucketId']` is NOT set?
If so, `list_buckets(bucket_name)` will recurse back to `get_bucket_by_name(bucket_name)`.

Let's see if we can avoid this recursion by using the cache or something. But if it's not in the cache, we have to ask the server.
The only way to ask the server for a bucket ID by name (when we don't have the ID) is to list buckets.

Wait, can we use `self.session.list_buckets(account_id, bucket_id=None)`? No, that's exactly what `list_buckets()` does, and that's what fails for restricted keys.

If we have a restricted key, we MUST use `bucket_id` in `list_buckets`.
But where do we get `bucket_id`? From `allowed['bucketId']`.

If `allowed['bucketName']` is set, but `allowed['bucketId']` is NOT set, then we are in a weird state. But according to the B2 API, they should be together.

Let's check if `B2Api.get_bucket_by_name` can be simplified.

If we change it to:
```python
    def get_bucket_by_name(self, bucket_name):
        # ...
        self.check_bucket_restrictions(bucket_name)
        # ...
        
        allowed = self.account_info.get_allowed()
        # If we have a restriction, we must pass bucket_name to list_buckets
        # to avoid RestrictedBucket exception.
        list_buckets_args = (bucket_name,) if allowed['bucketName'] is not None else ()
        
        for bucket in self.list_buckets(*list_buckets_args):
            if bucket.name == bucket_name:
                return bucket
        raise NonExistentBucket(bucket_name)
```

Wait, if `allowed['bucketName']` is NOT None, `list_buckets(bucket_name)` is called.
Inside `list_buckets(bucket_name)`:
If `allowed['bucketId']` is NOT None, it works.
If `allowed['bucketId']` is None, it calls `get_bucket_by_name(bucket_name)` -> RECURSION.

So we really depend on `allowed['bucketId']` being present if `allowed['bucketName']` is present.

Let's test the fix with the reproduction script.

{
  "command": "python3 - <<'PY'\nimport sys\nfrom unittest.mock import MagicMock\nfrom b2.api import B2Api, RestrictedBucket, Bucket, NonExistentBucket\n\nclass MockAccountInfo:\n    def __init__(self, allowed_bucket_name=None, allowed_bucket_id=None):\n        self.allowed = {'bucketName': allowed_bucket_name, 'bucketId': allowed_bucket_id}\n    def get_account_id(self):\n        return 'account_id'\n    def get_allowed(self):\n        return self.allowed\n\nclass MockCache:\n    def get_bucket_id_or_none_from_bucket_name(self, name):\n        return None\n    def save_bucket(self, bucket):\n        pass\n    def set_bucket_name_cache(self, buckets):\n        pass\n\nclass MockSession:\n    def list_buckets(self, account_id, bucket_id=None):\n        if bucket_id == 'my-bucket-id':\n            return [{'bucketName': 'my-bucket', 'bucketId': 'my-bucket-id'}]\n        elif bucket_id is None:\n            if 'restricted' in account_id: # simulate restriction\n                raise Exception(\"Unauthorized\")\n            return [{'bucketName': 'all-buckets', 'bucketId': 'all-id'}]\n        return []\n\ndef BucketFactory_from_api_response(api, response):\n    return [Bucket(api, b['bucketId'], name=b['bucketName']) for b in response]\n\n# Monkeypatch BucketFactory\nimport b2.api\nb2.api.BucketFactory.from_api_response = BucketFactory_from_api_response\n\ndef test_bucket_scoped_key():\n    # Setup a bucket-scoped key\n    account_info = MockAccountInfo(allowed_bucket_name='my-bucket', allowed_bucket_id='my-bucket-id')\n    cache = MockCache()\n    session = MockSession()\n    api = B2Api(account_info, cache, session, MagicMock())\n    \n    try:\n        api.get_bucket_by_name('my-bucket')\n        print(\"SUCCESS\")\n    except Exception as e:\n        print(f\"FAILED: {type(e).__name__}: {e}\")\n\ndef test_unrestricted_key():\n    # Setup an unrestricted key\n    account_info = MockAccountInfo(allowed_bucket_name=None, allowed_bucket_id=None)\n    cache = MockCache()\n    session = MockSession()\n    api = B2Api(account_info, cache, session, MagicMock())\n    \n    try:\n        api.get_bucket_by_name('all-buckets')\n        print(\"SUCCESS\")\n    except Exception as e:\n        print(f\"FAILED: {type(e).__name__}: {e}\")\n\nprint(\"Running bucket-scoped key test...\")\ntest_bucket_scoped_key()\nprint(\"Running unrestricted key test...\")\ntest_unrestricted_key()\nPY\n"
}

{
  "status": "ok",
  "stdout": "Running bucket-scoped key test...\nFAILED: RestrictedBucket: Application key is restricted to bucket: my-bucket\nRunning unrestricted key test...\nFAILED: AttributeError: 'MockAccountInfo' object has no attribute 'get_api_url'\n",
  "stderr": "",
  "exit_code": 0
}

Wait, I got `RestrictedBucket` again. This is because I didn't apply the fix to the code I'm running. I just ran a script against the existing `b2.api.B2Api`.

Let's apply the fix to `b2/api.py` first.

I will modify `get_bucket_by_name` in `b2/api.py`.

```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)
```

To:

```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)

        allowed = self.account_info.get_allowed()
        if allowed['bucketName'] is not None:
            for bucket in self.list_buckets(bucket_name):
                if bucket.name == bucket_name:
                    return bucket
        else:
            for bucket in self.list_buckets():
                if bucket.name == bucket_name:
                    return bucket
        raise NonExistentBucket(bucket_name)
```

Actually, I can make it even cleaner:

```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)

        allowed = self.account_info.get_allowed()
        search_bucket_name = bucket_name if allowed['bucketName'] is not None else None
        for bucket in self.list_buckets(search_bucket_name):
            if bucket.name == bucket_name:
                return bucket
        raise NonExistentBucket(bucket_name)
```

Wait, if `allowed['bucketName']` is not None, then `search_bucket_name` is `bucket_name`. Then `list_buckets(bucket_name)` is called.
If `allowed['bucketName']` is None, then `search_bucket_name` is `None`. Then `list_buckets()` is called.
This is exactly what I wanted.

Let's apply this.

{
  "filepath": "b2/api.py",
  "new_string": "        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        allowed = self.account_info.get_allowed()\n        search_bucket_name = bucket_name if allowed['bucketName'] is not None else None\n        for bucket in self.list_buckets(search_bucket_name):\n            if bucket.name == bucket_name:\n                return bucket\n        raise NonExistentBucket(bucket_name)",
  "old_string": "        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)"
}

{
  "status": "ok",
  "filepath": "b2/api.py",
  "occurrences": 1,
  "strategy": "exact",
  "diff": "--- a/b2/api.py\n+++ b/b2/api.py\n@@ -216,7 +216,9 @@\n         if id_ is not None:\n             return Bucket(self, id_, name=bucket_name)\n \n-        for bucket in self.list_buckets():\n+        allowed = self.account_info.get_allowed()\n+        search_bucket_name = bucket_name if allowed['bucketName'] is not None else None\n+        for bucket in self.list_buckets(search_bucket_name):\n             if bucket.name == bucket_name:\n                 return bucket\n         raise NonExistentBucket(bucket_name)\n",
  "is_truncated": false
}

Now let's run the reproduction script again. I'll need to fix the `MockAccountInfo` to include `get_api_url` (or whatever else `B2Api` needs) or just mock it properly.

Let's see what `B2Api` constructor does.

{
  "end_line": 100,
  "filepath": "b2/api.py",
  "start_line": 1
}

{
  "status": "ok",
  "filepath": "b2/api.py",
  "content": "######################################################################\n#\n# File: b2/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\nimport six\n\nfrom .account_info.sqlite_account_info import SqliteAccountInfo\nfrom .account_info.exception import MissingAccountData\nfrom .b2http import B2Http\nfrom .bucket import Bucket, BucketFactory\nfrom .cache import AuthInfoCache, DummyCache\nfrom .transferer import Transferer\nfrom .exception import NonExistentBucket, RestrictedBucket\nfrom .file_version import FileVersionInfoFactory, FileIdAndName\nfrom .part import PartFactory\nfrom .raw_api import B2RawApi\nfrom .session import B2Session\nfrom .utils import B2TraceMeta, b2_url_encode, limit_trace_arguments\n\ntry:\n    import concurrent.futures as futures\nexcept ImportError:\n    import futures\n\n\ndef url_for_api(info, api_name):\n    if api_name in ['b2_download_file_by_id']:\n        base = info.get_download_url()\n    else:\n        base = info.get_api_url()\n    return base + '/b2api/v1/' + api_name\n\n\n@six.add_metaclass(B2TraceMeta)\nclass B2Api(object):\n    \"\"\"\n    Provides file-level access to B2 services.\n\n    While B2RawApi provides direct access to the B2 web APIs, this\n    class handles several things that simplify the task of uploading\n    and downloading files:\n      - re-acquires authorization tokens when they expire\n      - retrying uploads when an upload URL is busy\n      - breaking large files into parts\n      - emulating a directory structure (B2 buckets are flat)\n\n    Adds an object-oriented layer on top of the raw API, so that\n    buckets and files returned are Python objects with accessor\n    methods.\n\n    Also,  keeps a cache of information needed to access the service,\n    such as auth tokens and upload URLs.\n    \"\"\"\n\n    def __init__(self, account_info=None, cache=None, raw_api=None, max_upload_workers=10):\n        \"\"\"\n        Initializes the API using the given account info.\n        :param account_info:\n        :param cache:\n        :param raw_api:\n        :return:\n        \"\"\"\n        self.raw_api = raw_api or B2RawApi(B2Http())\n        if account_info is None:\n            account_info = SqliteAccountInfo()\n            if cache is None:\n                cache = AuthInfoCache(account_info)\n        self.session = B2Session(self, self.raw_api)\n        self.transferer = Transferer(self.session, account_info)\n        self.account_info = account_info\n        if cache is None:\n            cache = DummyCache()\n        self.cache = cache\n        self.upload_executor = None\n        self.max_workers = max_upload_workers\n\n    def set_thread_pool_size(self, max_workers):\n        \"\"\"\n        Sets the size of the thread pool to use for uploads and downloads.\n\n        Must be called before any work starts, or the thread pool will get\n        the default size of 1.\n        \"\"\"\n        if self.upload_executor is not None:\n            raise Exception('thread pool already created')\n        self.max_workers = max_workers\n\n    def get_thread_pool(self):\n        \"\"\"\n        Returns the thread pool executor to use for uploads and downloads.\n        \"\"\"\n        if self.upload_executor is None:\n            self.upload_executor = futures.ThreadPoolExecutor(max_workers=self.max_workers)\n        return self.upload_executor",
  "start_line": 1,
  "end_line": 100,
  "total_lines": 374,
  "is_truncated": false
}