backblaze__b2_command_line_tool-488

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

{
  "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
}

{
  "status": "ok",
  "filepath": "b2/api.py",
  "content": "        if allowed_bucket_name is not None:\n            if allowed_bucket_name != bucket_name:\n                raise RestrictedBucket(allowed_bucket_name)",
  "start_line": 370,
  "end_line": 372,
  "total_lines": 372,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "b2/api.py",
  "content": "        \"\"\"\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": 360,
  "end_line": 372,
  "total_lines": 372,
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": "./test/stub_account_info.py:103:    def get_allowed(self):\n./b2/raw_simulator.py:90:    def get_allowed(self):\n./b2/account_info/abstract.py:118:    def get_allowed(self):\n./b2/account_info/sqlite_account_info.py:255:    def get_allowed(self):\n./b2/account_info/in_memory.py:112:    def get_allowed(self):\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "b2/account_info/abstract.py",
  "content": "\n    @abstractmethod\n    def get_minimum_part_size(self):\n        \"\"\"\n        :return: returns the minimum number of bytes in a part of a large file\n        \"\"\"\n\n    @abstractmethod\n    def get_allowed(self):\n        \"\"\"\n        An 'allowed' dict, as returned by b2_authorize_account.\n        Never None; for account info that was saved before 'allowed' existed,\n        returns DEFAULT_ALLOWED.\n        \"\"\"\n\n    @limit_trace_arguments(only=['self', 'api_url', 'download_url', 'minimum_part_size', 'realm'])\n    def set_auth_data(\n        self,\n        account_id,\n        auth_token,\n        api_url,",
  "start_line": 110,
  "end_line": 130,
  "total_lines": 213,
  "is_truncated": false
}

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

{
  "status": "ok",
  "filepath": "b2/account_info/sqlite_account_info.py",
  "content": "        return self._get_account_info_or_raise('realm')\n\n    def get_minimum_part_size(self):\n        return self._get_account_info_or_raise('minimum_part_size')\n\n    def get_allowed(self):\n        \"\"\"\n        The 'allowed\" column was not in the original schema, so it may be NULL.\n        \"\"\"\n        allowed_json = self._get_account_info_or_raise('allowed')\n        if allowed_json is None:\n            return self.DEFAULT_ALLOWED\n        else:\n            return json.loads(allowed_json)\n\n    def _get_account_info_or_raise(self, column_name):",
  "start_line": 250,
  "end_line": 265,
  "total_lines": 309,
  "is_truncated": false
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "",
  "details": {
    "stdout": "",
    "stderr": "",
    "exit_code": 1
  }
}

{
  "status": "error",
  "error_type": "FileWriteError",
  "error_message": "Path traversal detected: '/tmp/repro.py' escapes workspace root."
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "",
  "details": {
    "stdout": "",
    "stderr": "",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "stdout": "./b2/bucket.py:634:class BucketFactory(object):\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "/workspace/repro.py",
  "size": 1627
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "F.\n======================================================================\nFAIL: test_get_bucket_by_name_restricted_key (__main__.TestB2ApiRestrictedKey)\n----------------------------------------------------------------------\nTraceback (most recent call last):\n  File \"/workspace/repro.py\", line 34, in test_get_bucket_by_name_restricted_key\n    api.get_bucket_by_name('my-bucket')\nAssertionError: RestrictedBucket not raised\n\n----------------------------------------------------------------------\nRan 2 tests in 0.001s\n\nFAILED (failures=1)\n",
  "details": {
    "stdout": "",
    "stderr": "F.\n======================================================================\nFAIL: test_get_bucket_by_name_restricted_key (__main__.TestB2ApiRestrictedKey)\n----------------------------------------------------------------------\nTraceback (most recent call last):\n  File \"/workspace/repro.py\", line 34, in test_get_bucket_by_name_restricted_key\n    api.get_bucket_by_name('my-bucket')\nAssertionError: RestrictedBucket not raised\n\n----------------------------------------------------------------------\nRan 2 tests in 0.001s\n\nFAILED (failures=1)\n",
    "exit_code": 1
  }
}

{
  "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",
  "start_line": 200,
  "end_line": 300,
  "total_lines": 372,
  "is_truncated": false
}