如何用OAuth2令牌访问Azure Blob Storage？支持Python实现吗？

Hey there! It looks like the issue you're hitting is most likely related to how you're formatting the Authorization header when calling the Blob Storage API. Let's break down the fix and walk through a proper Python implementation to handle file list, upload, and download operations.

First: Fix the Authorization Header for REST API Calls #

The error message you're seeing happens because when using an OAuth2 access token to authenticate Blob Storage requests, you need to use the Bearer authentication scheme instead of the traditional Shared Key signature. Here's what a valid request should look like:

Correct REST API Request Example (using curl) #

curl -X GET "https://account_name.blob.core.windows.net/container_name?restype=container&comp=list" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -H "x-ms-version: 2020-04-08"

• Replace YOUR_ACCESS_TOKEN with the access_token value you received from the token endpoint.
• The x-ms-version header is mandatory—use a recent stable version (like 2020-04-08 or later) to ensure OAuth2 authentication is supported.

Quick Checks for REST Calls: #

• Your scope (https://storage.azure.com/user_impersonation) is correct—this allows the token to be used for user-impersonated access to Blob Storage.
• Double-check that the user linked to the token has the right RBAC permissions on the storage account/container (e.g., Storage Blob Data Contributor). Even a valid token will fail if permissions are missing.

Python Implementation (Recommended: Use Azure SDK) #

Manually calling the REST API works, but using the official Azure Storage Blob SDK for Python simplifies token handling—including automatic refresh if you have a refresh token. Here's how to set it up:

Step 1: Install Required Packages #

pip install azure-storage-blob azure-identity

Option 1: Use an Existing Access Token #

If you already have an access token from your token exchange, you can pass it directly to the BlobServiceClient:

from azure.storage.blob import BlobServiceClient

# Replace with your actual values
access_token = "YOUR_ACCESS_TOKEN"
account_url = "https://account_name.blob.core.windows.net"
container_name = "container_name"

# Initialize the Blob service client with your access token
blob_service_client = BlobServiceClient(account_url=account_url, credential=access_token)

# 1. List all blobs in the container
container_client = blob_service_client.get_container_client(container_name)
print("Listing blobs:")
for blob in container_client.list_blobs():
    print(f"- {blob.name}")

# 2. Upload a local file to the container
local_file_path = "path/to/your/local/document.txt"
remote_blob_name = "uploaded_document.txt"

with open(local_file_path, "rb") as data:
    container_client.upload_blob(name=remote_blob_name, data=data)
print(f"Successfully uploaded {local_file_path} to {remote_blob_name}")

# 3. Download a blob to your local machine
downloaded_file_path = "path/to/save/downloaded/document.txt"
blob_client = container_client.get_blob_client(remote_blob_name)

with open(downloaded_file_path, "wb") as download_file:
    download_file.write(blob_client.download_blob().readall())
print(f"Successfully downloaded {remote_blob_name} to {downloaded_file_path}")

Option 2: Automate Token Refresh (Using Refresh Token) #

Since you requested the offline_access scope, you should have a refresh_token from your token exchange. Use AuthorizationCodeCredential from azure-identity to automatically refresh the access token when it expires:

from azure.identity import AuthorizationCodeCredential
from azure.storage.blob import BlobServiceClient

# Replace with your actual values
tenant_id = "YOUR_TENANT_ID"
client_id = "YOUR_CLIENT_ID"
client_secret = "YOUR_CLIENT_SECRET"
authorization_code = "YOUR_AUTHORIZATION_CODE"
redirect_uri = "http://localhost/myapp/"
account_url = "https://account_name.blob.core.windows.net"
container_name = "container_name"

# Initialize credential (handles token refresh automatically)
credential = AuthorizationCodeCredential(
    tenant_id=tenant_id,
    client_id=client_id,
    client_secret=client_secret,
    authorization_code=authorization_code,
    redirect_uri=redirect_uri
)

# Create the Blob service client
blob_service_client = BlobServiceClient(account_url=account_url, credential=credential)

# Perform operations (same as Option 1)
container_client = blob_service_client.get_container_client(container_name)
# ... list, upload, download operations ...

Key Notes for Python Implementation: #

• The SDK takes care of formatting the Authorization header and setting the correct API version automatically—no manual work needed.
• Make sure the user has the necessary RBAC permissions (like Storage Blob Data Contributor) to perform the operations you need.

内容的提问来源于stack exchange，提问作者subin_john_